Add full AARS system - Attendance & Accident Record System
- FastAPI backend with SQLite database - React + TailwindCSS frontend - JWT authentication - Excel upload/download for attendance and accident records - Shift roster management with S7a/S7b split - Lateness/OT calculation based on shift times - Roster info API endpoints - Export API with token query parameter support - Docker compose deployment
This commit is contained in:
IT狗
@@ -0,0 +1,72 @@
|
||||
from fastapi import Depends, HTTPException, status, Query
|
||||
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
||||
from sqlalchemy.orm import Session
|
||||
from jose import JWTError, jwt
|
||||
from database import get_db, SECRET_KEY, ALGORITHM
|
||||
from models import User
|
||||
|
||||
security = HTTPBearer()
|
||||
|
||||
|
||||
async def get_current_user(
|
||||
credentials: HTTPAuthorizationCredentials = Depends(security),
|
||||
db: Session = Depends(get_db)
|
||||
) -> User:
|
||||
"""Validate JWT token and return current user."""
|
||||
token = credentials.credentials
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid or expired token",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
user_id_str = payload.get("sub")
|
||||
if user_id_str is None:
|
||||
raise credentials_exception
|
||||
try:
|
||||
user_id = int(user_id_str)
|
||||
except (ValueError, TypeError):
|
||||
raise credentials_exception
|
||||
except JWTError:
|
||||
raise credentials_exception
|
||||
|
||||
user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
|
||||
if user is None:
|
||||
raise credentials_exception
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_admin(current_user: User = Depends(get_current_user)) -> User:
|
||||
"""Ensure current user is an admin."""
|
||||
if current_user.role != "admin":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Admin access required"
|
||||
)
|
||||
return current_user
|
||||
|
||||
|
||||
async def get_current_user_from_query(
|
||||
token: str = Query(...),
|
||||
db: Session = Depends(get_db)
|
||||
) -> User:
|
||||
"""Validate JWT token from query parameter and return current user."""
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid or expired token",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
user_id_str = payload.get("sub")
|
||||
if user_id_str is None:
|
||||
raise credentials_exception
|
||||
user_id = int(user_id_str)
|
||||
except (JWTError, ValueError, TypeError):
|
||||
raise credentials_exception
|
||||
|
||||
user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
|
||||
if user is None:
|
||||
raise credentials_exception
|
||||
return user
|
||||
@@ -0,0 +1,72 @@
|
||||
import os
|
||||
from datetime import datetime, timezone
|
||||
from sqlalchemy import create_engine, Column, Integer, String, Text, Float, Boolean, DateTime, Date, Time, JSON
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from passlib.context import CryptContext
|
||||
from jose import jwt
|
||||
from typing import Optional, List
|
||||
from pydantic import BaseModel
|
||||
import secret
|
||||
|
||||
# Database setup
|
||||
DATABASE_URL = "sqlite:///./data/aars.db"
|
||||
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
|
||||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
Base = declarative_base()
|
||||
|
||||
pwd_context = CryptContext(schemes=["sha256_crypt"], deprecated="auto")
|
||||
|
||||
# JWT Settings
|
||||
SECRET_KEY = secret.JWT_SECRET if hasattr(secret, 'JWT_SECRET') else "aars-dev-secret-change-in-production"
|
||||
ALGORITHM = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_HOURS = 24
|
||||
|
||||
|
||||
def create_access_token(data: dict) -> str:
|
||||
expire = datetime.now(timezone.utc).timestamp() + (ACCESS_TOKEN_EXPIRE_HOURS * 3600)
|
||||
to_encode = data.copy()
|
||||
to_encode.update({"exp": expire})
|
||||
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||||
|
||||
|
||||
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||||
return pwd_context.verify(plain_password, hashed_password)
|
||||
|
||||
|
||||
def get_password_hash(password: str) -> str:
|
||||
return pwd_context.hash(password)
|
||||
|
||||
|
||||
def get_db():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def init_db():
|
||||
"""Initialize database with tables and default admin user."""
|
||||
Base.metadata.create_all(bind=engine)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
# Check if admin exists
|
||||
from models import User
|
||||
admin = db.query(User).filter(User.email == "admin@aars.hk").first()
|
||||
if not admin:
|
||||
# Create default admin
|
||||
admin = User(
|
||||
email="admin@aars.hk",
|
||||
password_hash=get_password_hash("admin123"),
|
||||
name="系統管理員",
|
||||
role="admin",
|
||||
is_active=True
|
||||
)
|
||||
db.add(admin)
|
||||
db.commit()
|
||||
print("✅ Default admin user created: admin@aars.hk / admin123")
|
||||
else:
|
||||
print("✅ Admin user already exists")
|
||||
finally:
|
||||
db.close()
|
||||
+781
@@ -0,0 +1,781 @@
|
||||
import os
|
||||
import secret
|
||||
from fastapi import FastAPI, Depends, HTTPException, Query, UploadFile, File, Form, Request
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import FileResponse, RedirectResponse
|
||||
from sqlalchemy.orm import Session
|
||||
from datetime import date, datetime, timedelta
|
||||
from typing import Optional, List
|
||||
import json
|
||||
|
||||
from database import get_db, create_access_token, verify_password, init_db, Base, engine
|
||||
from models import User
|
||||
from schemas import *
|
||||
from auth import get_current_user, get_current_admin, get_current_user_from_query
|
||||
|
||||
app = FastAPI(title="AARS API")
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
# Data directory for storing uploaded Excel files
|
||||
DATA_DIR = "/app/data"
|
||||
os.makedirs(DATA_DIR, exist_ok=True)
|
||||
|
||||
@app.on_event("startup")
|
||||
async def startup():
|
||||
init_db()
|
||||
|
||||
# ============ Auth ============
|
||||
@app.post("/api/auth/login", response_model=TokenResponse)
|
||||
async def login(form_data: LoginRequest, db: Session = Depends(get_db)):
|
||||
user = db.query(User).filter(User.email == form_data.email).first()
|
||||
if not user or not verify_password(form_data.password, user.password_hash):
|
||||
raise HTTPException(status_code=401, detail="Invalid email or password")
|
||||
access_token = create_access_token({"sub": str(user.id)})
|
||||
return TokenResponse(access_token=access_token)
|
||||
|
||||
@app.get("/api/auth/me", response_model=UserResponse)
|
||||
async def get_me(current_user: User = Depends(get_current_user)):
|
||||
return current_user
|
||||
|
||||
# ============ Excel File Storage ============
|
||||
def get_excel_path(section: str) -> str:
|
||||
"""Get path to the stored Excel file for a section"""
|
||||
return os.path.join(DATA_DIR, f"{section}.xlsx")
|
||||
|
||||
def read_excel_file(section: str):
|
||||
"""Read Excel file and return headers and rows"""
|
||||
import openpyxl
|
||||
path = get_excel_path(section)
|
||||
if not os.path.exists(path):
|
||||
return None, []
|
||||
|
||||
wb = openpyxl.load_workbook(path, data_only=True)
|
||||
ws = wb.active
|
||||
|
||||
headers = [cell.value for cell in ws[1]]
|
||||
rows = list(ws.iter_rows(min_row=2, values_only=True))
|
||||
|
||||
return headers, rows
|
||||
|
||||
# ============ Upload Excel (Replace entire file) ============
|
||||
@app.delete("/api/upload/{section}")
|
||||
async def delete_uploaded_file(
|
||||
section: str,
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Delete the stored Excel file for a section"""
|
||||
if section not in ["attendance", "accident"]:
|
||||
raise HTTPException(status_code=400, detail="Invalid section")
|
||||
|
||||
path = get_excel_path(section)
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
return {"message": f"{section} file deleted"}
|
||||
|
||||
@app.post("/api/upload/{section}")
|
||||
async def upload_excel(
|
||||
section: str,
|
||||
file: UploadFile = File(...),
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Upload Excel file - replaces any existing file for this section"""
|
||||
if section not in ["attendance", "accident"]:
|
||||
raise HTTPException(status_code=400, detail="Invalid section")
|
||||
|
||||
if not file.filename:
|
||||
raise HTTPException(status_code=400, detail="No filename provided")
|
||||
|
||||
# Save the file
|
||||
path = get_excel_path(section)
|
||||
contents = await file.read()
|
||||
|
||||
with open(path, "wb") as f:
|
||||
f.write(contents)
|
||||
|
||||
# Return info about the uploaded file
|
||||
headers, rows = read_excel_file(section)
|
||||
|
||||
return {
|
||||
"message": f"{section} file uploaded successfully",
|
||||
"filename": file.filename,
|
||||
"rows": len(rows),
|
||||
"columns": len(headers),
|
||||
"headers": headers
|
||||
}
|
||||
|
||||
|
||||
# ============ Attendance Calculation ============
|
||||
def get_shift_schedule_full(shift_code: str, weekday: str) -> Optional[tuple]:
|
||||
"""Get the expected start and end time for a shift on a given weekday.
|
||||
Returns (start_time, end_time) tuple.
|
||||
"""
|
||||
roster_path = get_excel_path("roster")
|
||||
if not os.path.exists(roster_path):
|
||||
return None
|
||||
|
||||
import openpyxl
|
||||
wb = openpyxl.load_workbook(roster_path, data_only=True)
|
||||
ws = wb.active
|
||||
|
||||
headers = [cell.value for cell in ws[1]]
|
||||
|
||||
# Find column indices
|
||||
shift_col = None
|
||||
for i, h in enumerate(headers):
|
||||
if h == '班次':
|
||||
shift_col = i
|
||||
break
|
||||
|
||||
if shift_col is None:
|
||||
return None
|
||||
|
||||
# Weekday mapping
|
||||
weekday_map = {
|
||||
'Monday': '星期一', 'Tuesday': '星期二', 'Wednesday': '星期三',
|
||||
'Thursday': '星期四', 'Friday': '星期五', 'Saturday': '星期六',
|
||||
'Sunday': '星期日', '星期一': '星期一', '星期二': '星期二',
|
||||
'星期三': '星期三', '星期四': '星期四', '星期五': '星期五',
|
||||
'星期六': '星期六', '星期日': '星期日'
|
||||
}
|
||||
|
||||
day_col_map = {
|
||||
'星期一': None, '星期二': None, '星期三': None,
|
||||
'星期四': None, '星期五': None, '星期六': None, '星期日': None
|
||||
}
|
||||
|
||||
for i, h in enumerate(headers):
|
||||
if h in day_col_map:
|
||||
day_col_map[h] = i
|
||||
|
||||
target_day = weekday_map.get(weekday, weekday)
|
||||
|
||||
# Find the shift row
|
||||
for row in ws.iter_rows(min_row=2, values_only=True):
|
||||
if row[shift_col] == shift_code:
|
||||
day_col = day_col_map.get(target_day)
|
||||
if day_col is not None:
|
||||
schedule = row[day_col]
|
||||
if schedule and schedule != '-' and schedule != '休息':
|
||||
# Parse "HH:MM-HH:MM" format
|
||||
try:
|
||||
parts = str(schedule).split('-')
|
||||
if len(parts) == 2:
|
||||
start_str = parts[0].strip()
|
||||
end_str = parts[1].strip()
|
||||
start_time = datetime.strptime(start_str, '%H:%M').time()
|
||||
end_time = datetime.strptime(end_str, '%H:%M').time()
|
||||
return (start_time, end_time)
|
||||
except:
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
def parse_datetime(time_val) -> Optional[datetime]:
|
||||
"""Parse datetime from various formats"""
|
||||
if not time_val:
|
||||
return None
|
||||
if isinstance(time_val, datetime):
|
||||
return time_val
|
||||
if isinstance(time_val, str):
|
||||
for fmt in ['%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M', '%H:%M:%S', '%H:%M']:
|
||||
try:
|
||||
return datetime.strptime(time_val, fmt)
|
||||
except:
|
||||
continue
|
||||
return None
|
||||
|
||||
def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
|
||||
"""Calculate attendance status: late_minutes, early_minutes, ot_minutes, is_missing
|
||||
|
||||
Returns: {
|
||||
"is_missing": bool,
|
||||
"late_minutes": int,
|
||||
"early_minutes": int,
|
||||
"ot_minutes": int,
|
||||
"status_code": str, # e.g. "late_early", "early_ot", "normal"
|
||||
"status_text": str,
|
||||
"expected_in": str,
|
||||
"expected_out": str,
|
||||
"actual_in": str,
|
||||
"actual_out": str
|
||||
}
|
||||
"""
|
||||
result = {
|
||||
"is_missing": False,
|
||||
"late_minutes": 0,
|
||||
"early_minutes": 0,
|
||||
"ot_minutes": 0,
|
||||
"status_code": "normal",
|
||||
"status_text": "正常",
|
||||
"expected_in": shift_start.strftime("%H:%M") if shift_start else "-",
|
||||
"expected_out": shift_end.strftime("%H:%M") if shift_end else "-",
|
||||
"actual_in": "-",
|
||||
"actual_out": "-"
|
||||
}
|
||||
|
||||
# Check for missing punch
|
||||
if not check_in or not check_out:
|
||||
result["is_missing"] = True
|
||||
result["status_code"] = "missing"
|
||||
result["status_text"] = "缺勤"
|
||||
if check_in:
|
||||
result["actual_in"] = check_in.strftime("%H:%M") if isinstance(check_in, datetime) else str(check_in)
|
||||
if check_out:
|
||||
result["actual_out"] = check_out.strftime("%H:%M") if isinstance(check_out, datetime) else str(check_out)
|
||||
return result
|
||||
|
||||
# Parse check-in/out times
|
||||
check_in_dt = parse_datetime(check_in)
|
||||
check_out_dt = parse_datetime(check_out)
|
||||
|
||||
if not check_in_dt or not check_out_dt:
|
||||
result["is_missing"] = True
|
||||
result["status_code"] = "missing"
|
||||
result["status_text"] = "缺勤"
|
||||
return result
|
||||
|
||||
result["actual_in"] = check_in_dt.strftime("%H:%M")
|
||||
result["actual_out"] = check_out_dt.strftime("%H:%M")
|
||||
|
||||
# Skip if invalid times (Excel default "0" = 18:00)
|
||||
if check_in_dt.hour == 18 and check_in_dt.minute == 0 and check_in_dt.second == 0:
|
||||
result["is_missing"] = True
|
||||
result["status_code"] = "missing"
|
||||
result["status_text"] = "缺勤"
|
||||
return result
|
||||
|
||||
if not shift_start or not shift_end:
|
||||
# No roster data, can't calculate
|
||||
return result
|
||||
|
||||
# Calculate late minutes
|
||||
check_in_time = check_in_dt.time()
|
||||
if check_in_time > shift_start:
|
||||
diff = (datetime.combine(check_in_dt.date(), check_in_time) -
|
||||
datetime.combine(check_in_dt.date(), shift_start)).total_seconds() / 60
|
||||
result["late_minutes"] = int(diff)
|
||||
|
||||
# Calculate early minutes (leaving before scheduled end)
|
||||
check_out_time = check_out_dt.time()
|
||||
if shift_end > shift_start:
|
||||
# Handle overnight shifts
|
||||
if shift_end < shift_start:
|
||||
# Assume next day
|
||||
expected_end_dt = datetime.combine(check_out_dt.date() + timedelta(days=1), shift_end)
|
||||
else:
|
||||
expected_end_dt = datetime.combine(check_out_dt.date(), shift_end)
|
||||
|
||||
if check_out_dt < expected_end_dt:
|
||||
diff = (expected_end_dt - check_out_dt).total_seconds() / 60
|
||||
result["early_minutes"] = int(diff)
|
||||
|
||||
# Calculate OT minutes (leaving after scheduled end)
|
||||
if check_out_dt > expected_end_dt:
|
||||
diff = (check_out_dt - expected_end_dt).total_seconds() / 60
|
||||
result["ot_minutes"] = int(diff)
|
||||
|
||||
# Determine status code and text
|
||||
status_parts = []
|
||||
if result["late_minutes"] > 0:
|
||||
status_parts.append(f"遲到{result['late_minutes']}分")
|
||||
if result["early_minutes"] > 0:
|
||||
status_parts.append(f"早退{result['early_minutes']}分")
|
||||
if result["ot_minutes"] > 0:
|
||||
status_parts.append(f"OT{result['ot_minutes']}分")
|
||||
|
||||
if result["late_minutes"] > 0 and result["early_minutes"] > 0:
|
||||
result["status_code"] = "late_early"
|
||||
elif result["late_minutes"] > 0 and result["ot_minutes"] > 0:
|
||||
result["status_code"] = "late_ot"
|
||||
elif result["early_minutes"] > 0 and result["ot_minutes"] > 0:
|
||||
result["status_code"] = "early_ot"
|
||||
elif result["late_minutes"] > 0:
|
||||
result["status_code"] = "late"
|
||||
elif result["early_minutes"] > 0:
|
||||
result["status_code"] = "early"
|
||||
elif result["ot_minutes"] > 0:
|
||||
result["status_code"] = "ot"
|
||||
else:
|
||||
result["status_code"] = "normal"
|
||||
|
||||
result["status_text"] = " / ".join(status_parts) if status_parts else "正常"
|
||||
|
||||
|
||||
@app.get("/api/attendance/lateness")
|
||||
async def get_lateness_stats(current_user: User = Depends(get_current_user)):
|
||||
"""Calculate full attendance statistics: late, early, OT"""
|
||||
headers, rows = read_excel_file("attendance")
|
||||
if headers is None or not rows:
|
||||
return {"stats": [], "total_late": 0, "total_early": 0, "total_ot": 0, "total_missing": 0}
|
||||
|
||||
staff_col = None; date_col = None; week_col = None; checkin_col = None; checkout_col = None; shift_col = None
|
||||
|
||||
for i, h in enumerate(headers):
|
||||
h_lower = str(h).lower() if h else ''
|
||||
if 'staff' in h_lower or '員工' in h_lower or 'name' in h_lower:
|
||||
staff_col = i
|
||||
elif 'date' in h_lower or '日期' in h_lower:
|
||||
date_col = i
|
||||
elif 'week' in h_lower or '星期' in h_lower:
|
||||
week_col = i
|
||||
elif 'check' in h_lower and 'in' in h_lower or '上班' in h_lower:
|
||||
checkin_col = i
|
||||
elif 'check' in h_lower and 'out' in h_lower or '下班' in h_lower:
|
||||
checkout_col = i
|
||||
elif '班次' in h_lower:
|
||||
shift_col = i
|
||||
|
||||
if staff_col is None or checkin_col is None or shift_col is None:
|
||||
return {"error": "Missing required columns", "headers": headers}
|
||||
|
||||
employee_stats = {}
|
||||
totals = {"late": 0, "early": 0, "ot": 0, "missing": 0}
|
||||
|
||||
for row in rows:
|
||||
staff = row[staff_col]; check_in = row[checkin_col]
|
||||
check_out = row[checkout_col] if checkout_col is not None else None
|
||||
shift_code = row[shift_col]; week_day = row[week_col] if week_col is not None else None
|
||||
|
||||
if not staff or not shift_code:
|
||||
continue
|
||||
|
||||
shift_times = get_shift_schedule_full(shift_code, week_day or '')
|
||||
shift_start = shift_times[0] if shift_times else None
|
||||
shift_end = shift_times[1] if shift_times else None
|
||||
|
||||
status = calculate_attendance_status(check_in, check_out, shift_start, shift_end)
|
||||
|
||||
staff_key = str(staff)
|
||||
if staff_key not in employee_stats:
|
||||
employee_stats[staff_key] = {"name": staff, "shift": shift_code, "late_count": 0, "late_minutes": 0, "early_count": 0, "early_minutes": 0, "ot_count": 0, "ot_minutes": 0, "missing_count": 0}
|
||||
|
||||
if status["is_missing"]:
|
||||
employee_stats[staff_key]["missing_count"] += 1
|
||||
totals["missing"] += 1
|
||||
else:
|
||||
if status["late_minutes"] > 0:
|
||||
employee_stats[staff_key]["late_count"] += 1
|
||||
employee_stats[staff_key]["late_minutes"] += status["late_minutes"]
|
||||
totals["late"] += status["late_minutes"]
|
||||
if status["early_minutes"] > 0:
|
||||
employee_stats[staff_key]["early_count"] += 1
|
||||
employee_stats[staff_key]["early_minutes"] += status["early_minutes"]
|
||||
totals["early"] += status["early_minutes"]
|
||||
if status["ot_minutes"] > 0:
|
||||
employee_stats[staff_key]["ot_count"] += 1
|
||||
employee_stats[staff_key]["ot_minutes"] += status["ot_minutes"]
|
||||
totals["ot"] += status["ot_minutes"]
|
||||
|
||||
stats = sorted(employee_stats.values(), key=lambda x: x["late_minutes"], reverse=True)
|
||||
return {"stats": stats, **totals}
|
||||
|
||||
@app.get("/api/attendance/lateness/records")
|
||||
async def get_lateness_records(
|
||||
staff: Optional[str] = None,
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Get detailed attendance status records"""
|
||||
headers, rows = read_excel_file("attendance")
|
||||
if headers is None or not rows:
|
||||
return []
|
||||
|
||||
staff_col = None; date_col = None; week_col = None; checkin_col = None; checkout_col = None; shift_col = None
|
||||
|
||||
for i, h in enumerate(headers):
|
||||
h_lower = str(h).lower() if h else ''
|
||||
if 'staff' in h_lower or '員工' in h_lower or 'name' in h_lower:
|
||||
staff_col = i
|
||||
elif 'date' in h_lower or '日期' in h_lower:
|
||||
date_col = i
|
||||
elif 'week' in h_lower or '星期' in h_lower:
|
||||
week_col = i
|
||||
elif 'check' in h_lower and 'in' in h_lower or '上班' in h_lower:
|
||||
checkin_col = i
|
||||
elif 'check' in h_lower and 'out' in h_lower or '下班' in h_lower:
|
||||
checkout_col = i
|
||||
elif '班次' in h_lower:
|
||||
shift_col = i
|
||||
|
||||
records = []
|
||||
for row in rows:
|
||||
staff_val = row[staff_col] if staff_col is not None else None
|
||||
check_in = row[checkin_col] if checkin_col is not None else None
|
||||
check_out = row[checkout_col] if checkout_col is not None else None
|
||||
shift_code = row[shift_col] if shift_col is not None else None
|
||||
week_day = row[week_col] if week_col is not None else None
|
||||
date_val = row[date_col] if date_col is not None else None
|
||||
|
||||
if staff_val and shift_code:
|
||||
shift_times = get_shift_schedule_full(shift_code, week_day or '')
|
||||
shift_start = shift_times[0] if shift_times else None
|
||||
shift_end = shift_times[1] if shift_times else None
|
||||
|
||||
status = calculate_attendance_status(check_in, check_out, shift_start, shift_end)
|
||||
|
||||
record = {
|
||||
"staff": staff_val, "date": str(date_val)[:10] if date_val else None,
|
||||
"weekday": week_day, "shift": shift_code,
|
||||
"status_code": status["status_code"], "status_text": status["status_text"],
|
||||
"expected_in": status["expected_in"], "expected_out": status["expected_out"],
|
||||
"actual_in": status["actual_in"], "actual_out": status["actual_out"],
|
||||
"late_minutes": status["late_minutes"], "early_minutes": status["early_minutes"], "ot_minutes": status["ot_minutes"]
|
||||
}
|
||||
|
||||
if staff is None or record["staff"] == staff:
|
||||
records.append(record)
|
||||
|
||||
return records
|
||||
|
||||
@app.delete("/api/roster")
|
||||
async def delete_roster(current_user: User = Depends(get_current_user)):
|
||||
"""Delete the stored roster file"""
|
||||
path = get_excel_path("roster")
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
return {"message": "Roster file deleted"}
|
||||
|
||||
@app.get("/api/roster/info")
|
||||
async def get_roster_info(current_user: User = Depends(get_current_user)):
|
||||
"""Get info about stored roster Excel file"""
|
||||
path = get_excel_path("roster")
|
||||
if not os.path.exists(path):
|
||||
return {"uploaded": False, "rows": 0, "columns": 0, "headers": []}
|
||||
|
||||
headers, rows = read_excel_file("roster")
|
||||
return {
|
||||
"uploaded": True,
|
||||
"rows": len(rows) if rows else 0,
|
||||
"columns": len(headers) if headers else 0,
|
||||
"headers": headers if headers else []
|
||||
}
|
||||
|
||||
@app.post("/api/roster/upload")
|
||||
async def upload_roster(
|
||||
file: UploadFile = File(...),
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Upload Roster Excel file"""
|
||||
if not file.filename:
|
||||
raise HTTPException(status_code=400, detail="No filename provided")
|
||||
|
||||
path = get_excel_path("roster")
|
||||
contents = await file.read()
|
||||
|
||||
with open(path, "wb") as f:
|
||||
f.write(contents)
|
||||
|
||||
headers, rows = read_excel_file("roster")
|
||||
|
||||
return {
|
||||
"message": "Roster file uploaded successfully",
|
||||
"filename": file.filename,
|
||||
"rows": len(rows) if rows else 0,
|
||||
"columns": len(headers) if headers else 0,
|
||||
"headers": headers if headers else []
|
||||
}
|
||||
|
||||
@app.get("/api/roster/data")
|
||||
async def get_roster_data(current_user: User = Depends(get_current_user)):
|
||||
"""Get full roster Excel data"""
|
||||
headers, rows = read_excel_file("roster")
|
||||
if headers is None:
|
||||
return {"headers": [], "rows": []}
|
||||
return {"headers": headers, "rows": rows}
|
||||
|
||||
@app.get("/api/roster/shifts")
|
||||
async def get_shifts(current_user: User = Depends(get_current_user)):
|
||||
"""Get all shifts from roster file"""
|
||||
headers, rows = read_excel_file("roster")
|
||||
if headers is None:
|
||||
return []
|
||||
|
||||
# Find columns
|
||||
shift_col = None
|
||||
desc_col = None
|
||||
for i, h in enumerate(headers):
|
||||
if h and str(h).lower() in ['班次', 'shift', 'shift code']:
|
||||
shift_col = i
|
||||
elif h and str(h).lower() in ['描述', 'description', 'desc']:
|
||||
desc_col = i
|
||||
|
||||
shifts = []
|
||||
for row in rows:
|
||||
shift_code = row[shift_col] if shift_col is not None else None
|
||||
if shift_code:
|
||||
shifts.append({
|
||||
"code": shift_code,
|
||||
"description": row[desc_col] if desc_col is not None else ""
|
||||
})
|
||||
|
||||
return shifts
|
||||
|
||||
@app.get("/api/roster/assignments")
|
||||
async def get_assignments(current_user: User = Depends(get_current_user)):
|
||||
"""Get employee-shift assignments"""
|
||||
# Read from attendance data
|
||||
headers, rows = read_excel_file("attendance")
|
||||
if headers is None:
|
||||
return {}
|
||||
|
||||
# Find 班次 column
|
||||
shift_col = None
|
||||
emp_col = None
|
||||
for i, h in enumerate(headers):
|
||||
if h and '班次' in str(h):
|
||||
shift_col = i
|
||||
elif h and any(kw in str(h).lower() for kw in ['員工', 'employee', 'name', '員工名稱']):
|
||||
emp_col = i
|
||||
|
||||
assignments = {}
|
||||
if shift_col is not None:
|
||||
for row in rows:
|
||||
emp = row[emp_col] if emp_col is not None and emp_col < len(row) else None
|
||||
shift = row[shift_col] if shift_col is not None and shift_col < len(row) else None
|
||||
if emp and shift:
|
||||
assignments[str(emp)] = str(shift)
|
||||
|
||||
return assignments
|
||||
|
||||
@app.post("/api/roster/assign")
|
||||
async def assign_shift(
|
||||
employee: str = Form(...),
|
||||
shift: str = Form(...),
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Assign employee to shift"""
|
||||
assignments_path = os.path.join(DATA_DIR, "shift_assignments.json")
|
||||
|
||||
# Load existing
|
||||
if os.path.exists(assignments_path):
|
||||
with open(assignments_path, "r") as f:
|
||||
assignments = json.load(f)
|
||||
else:
|
||||
assignments = {}
|
||||
|
||||
assignments[employee] = shift
|
||||
|
||||
return RedirectResponse(url="/")
|
||||
@app.get("/api/{section}/info")
|
||||
async def get_section_info(section: str):
|
||||
"""Get info about stored Excel file"""
|
||||
if section not in ["attendance", "accident"]:
|
||||
raise HTTPException(status_code=400, detail="Invalid section")
|
||||
|
||||
headers, rows = read_excel_file(section)
|
||||
if headers is None:
|
||||
return {"uploaded": False, "rows": 0, "columns": 0, "headers": []}
|
||||
|
||||
return {
|
||||
"uploaded": True,
|
||||
"rows": len(rows),
|
||||
"columns": len(headers),
|
||||
"headers": headers
|
||||
}
|
||||
|
||||
# ============ List Records (from Excel) ============
|
||||
@app.get("/api/{section}")
|
||||
async def list_records(
|
||||
section: str,
|
||||
page: int = Query(1, ge=1),
|
||||
page_size: int = Query(50, ge=1, le=1000),
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""List records from stored Excel file"""
|
||||
if section not in ["attendance", "accident"]:
|
||||
raise HTTPException(status_code=400, detail="Invalid section")
|
||||
|
||||
headers, rows = read_excel_file(section)
|
||||
if headers is None:
|
||||
return {"total": 0, "page": page, "page_size": page_size, "data": []}
|
||||
|
||||
# Pagination
|
||||
start = (page - 1) * page_size
|
||||
end = start + page_size
|
||||
page_rows = rows[start:end]
|
||||
|
||||
# Build response with row number as id
|
||||
data = []
|
||||
for idx, row in enumerate(page_rows, start=start + 1):
|
||||
record = {"_row": idx} # Row number (1-based, excluding header)
|
||||
for i, header in enumerate(headers):
|
||||
if header:
|
||||
record[str(header)] = row[i] if i < len(row) else None
|
||||
data.append(record)
|
||||
|
||||
return {
|
||||
"total": len(rows),
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"data": data
|
||||
}
|
||||
|
||||
# ============ Dashboard (from Excel) ============
|
||||
@app.get("/api/dashboard/{section}")
|
||||
async def dashboard(section: str, current_user: User = Depends(get_current_user)):
|
||||
"""Get dashboard stats from stored Excel file"""
|
||||
if section not in ["attendance", "accident"]:
|
||||
raise HTTPException(status_code=400, detail="Invalid section")
|
||||
|
||||
headers, rows = read_excel_file(section)
|
||||
if headers is None or not rows:
|
||||
return {"total": 0, "this_month": 0, "by_status": {}, "by_department": {}, "trend": []}
|
||||
|
||||
today = date.today()
|
||||
month_start = date(today.year, today.month, 1)
|
||||
|
||||
# Find date column (look for common date column names)
|
||||
date_col_idx = None
|
||||
for i, h in enumerate(headers):
|
||||
if h and any(kw in str(h).lower() for kw in ['date', '日期', '時間']):
|
||||
date_col_idx = i
|
||||
break
|
||||
|
||||
# Find department column
|
||||
dept_col_idx = None
|
||||
for i, h in enumerate(headers):
|
||||
if h and any(kw in str(h).lower() for kw in ['dept', '部門', '部門名稱']):
|
||||
dept_col_idx = i
|
||||
break
|
||||
|
||||
total = len(rows)
|
||||
this_month = 0
|
||||
by_department = {}
|
||||
trend = []
|
||||
|
||||
for row in rows:
|
||||
# Count this month
|
||||
if date_col_idx is not None and date_col_idx < len(row):
|
||||
val = row[date_col_idx]
|
||||
if val:
|
||||
row_date = None
|
||||
if isinstance(val, datetime):
|
||||
row_date = val.date()
|
||||
elif isinstance(val, date):
|
||||
row_date = val
|
||||
elif isinstance(val, str):
|
||||
for fmt in ['%Y-%m-%d', '%d/%m/%Y', '%m/%d/%Y']:
|
||||
try:
|
||||
row_date = datetime.strptime(val, fmt).date()
|
||||
break
|
||||
except:
|
||||
pass
|
||||
if row_date and row_date >= month_start:
|
||||
this_month += 1
|
||||
|
||||
# Count by department
|
||||
if dept_col_idx is not None and dept_col_idx < len(row):
|
||||
dept = str(row[dept_col_idx] or 'Unknown')
|
||||
by_department[dept] = by_department.get(dept, 0) + 1
|
||||
|
||||
return {
|
||||
"total": total,
|
||||
"today": 0,
|
||||
"this_month": this_month,
|
||||
"by_status": {},
|
||||
"by_department": by_department,
|
||||
"trend": []
|
||||
}
|
||||
|
||||
# ============ Record Detail (from Excel) ============
|
||||
@app.get("/api/{section}/{row_id}")
|
||||
async def get_record(
|
||||
section: str,
|
||||
row_id: int,
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Get a specific record by row number"""
|
||||
if section not in ["attendance", "accident"]:
|
||||
raise HTTPException(status_code=400, detail="Invalid section")
|
||||
|
||||
headers, rows = read_excel_file(section)
|
||||
if headers is None:
|
||||
raise HTTPException(status_code=404, detail="No data file found")
|
||||
|
||||
idx = row_id - 1 # Convert to 0-based index
|
||||
if idx < 0 or idx >= len(rows):
|
||||
raise HTTPException(status_code=404, detail="Record not found")
|
||||
|
||||
row = rows[idx]
|
||||
record = {"_row": row_id}
|
||||
for i, header in enumerate(headers):
|
||||
if header:
|
||||
record[str(header)] = row[i] if i < len(row) else None
|
||||
|
||||
return record
|
||||
|
||||
# ============ Export ============
|
||||
@app.get("/api/export/{section}/excel")
|
||||
async def export_excel(
|
||||
request: Request,
|
||||
section: str,
|
||||
token: Optional[str] = Query(None),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""Export the stored Excel file"""
|
||||
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
||||
from jose import JWTError, jwt
|
||||
from database import SECRET_KEY, ALGORITHM
|
||||
|
||||
credentials_exception = HTTPException(
|
||||
status_code=401,
|
||||
detail="Invalid or expired token",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
current_user = None
|
||||
|
||||
# If token in query, use it
|
||||
if token:
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
user_id = int(payload.get("sub"))
|
||||
current_user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
|
||||
except (JWTError, ValueError, TypeError):
|
||||
pass
|
||||
|
||||
# If no query token, try Authorization header
|
||||
if not current_user:
|
||||
auth_header = request.headers.get("Authorization", "")
|
||||
if auth_header.startswith("Bearer "):
|
||||
bearer_token = auth_header[7:]
|
||||
try:
|
||||
payload = jwt.decode(bearer_token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
user_id = int(payload.get("sub"))
|
||||
current_user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
|
||||
except (JWTError, ValueError, TypeError):
|
||||
pass
|
||||
|
||||
if not current_user:
|
||||
raise credentials_exception
|
||||
|
||||
if section not in ["attendance", "accident"]:
|
||||
raise HTTPException(status_code=400, detail="Invalid section")
|
||||
|
||||
path = get_excel_path(section)
|
||||
if not os.path.exists(path):
|
||||
raise HTTPException(status_code=404, detail="No file found")
|
||||
|
||||
return FileResponse(
|
||||
path,
|
||||
filename=f"{section}_export.xlsx",
|
||||
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
headers={"Content-Disposition": f"attachment; filename={section}_export.xlsx"}
|
||||
)
|
||||
|
||||
# ============ Roster / Shift Management ============
|
||||
|
||||
@app.get("/")
|
||||
async def root():
|
||||
return RedirectResponse(url="/index.html")
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
@@ -0,0 +1,94 @@
|
||||
from sqlalchemy import Column, Integer, String, Text, Float, Boolean, DateTime, Date, Time, JSON, ForeignKey
|
||||
from sqlalchemy.orm import relationship
|
||||
from sqlalchemy.sql import func
|
||||
from database import Base
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
def now_hkt():
|
||||
"""Return current time in HKT timezone."""
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
class User(Base):
|
||||
__tablename__ = "users"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
email = Column(String(255), unique=True, index=True, nullable=False)
|
||||
password_hash = Column(String(255), nullable=False)
|
||||
name = Column(String(255), nullable=False)
|
||||
role = Column(String(50), default="user") # admin, user
|
||||
is_active = Column(Boolean, default=True)
|
||||
created_at = Column(DateTime, default=now_hkt)
|
||||
updated_at = Column(DateTime, default=now_hkt, onupdate=now_hkt)
|
||||
deleted_at = Column(DateTime, nullable=True)
|
||||
|
||||
|
||||
class Attendance(Base):
|
||||
__tablename__ = "attendance"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
employee_id = Column(String(100), index=True, nullable=False)
|
||||
employee_name = Column(String(255), nullable=False)
|
||||
department = Column(String(100), index=True)
|
||||
date = Column(Date, nullable=False)
|
||||
check_in = Column(Time, nullable=True)
|
||||
check_out = Column(Time, nullable=True)
|
||||
overtime_hours = Column(Float, default=0)
|
||||
leave_type = Column(String(50), nullable=True) # annual, sick, unpaid, other
|
||||
leave_hours = Column(Float, default=0)
|
||||
status = Column(String(50), nullable=False) # present, absent, late, leave
|
||||
remarks = Column(Text, nullable=True)
|
||||
custom_fields = Column(JSON, default={})
|
||||
created_by = Column(Integer, ForeignKey("users.id"))
|
||||
created_at = Column(DateTime, default=now_hkt)
|
||||
updated_at = Column(DateTime, default=now_hkt, onupdate=now_hkt)
|
||||
deleted_at = Column(DateTime, nullable=True) # Soft delete
|
||||
|
||||
|
||||
class Accident(Base):
|
||||
__tablename__ = "accidents"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
date = Column(Date, nullable=False)
|
||||
time = Column(Time, nullable=True)
|
||||
location = Column(String(255), nullable=False)
|
||||
employee_name = Column(String(255), nullable=False)
|
||||
employee_id = Column(String(100))
|
||||
department = Column(String(100))
|
||||
description = Column(Text, nullable=False)
|
||||
severity = Column(String(50), nullable=False) # minor, moderate, serious, fatal
|
||||
medical_report = Column(Text, nullable=True)
|
||||
action_taken = Column(Text, nullable=True)
|
||||
responsible_person = Column(String(255), nullable=True)
|
||||
custom_fields = Column(JSON, default={})
|
||||
created_by = Column(Integer, ForeignKey("users.id"))
|
||||
created_at = Column(DateTime, default=now_hkt)
|
||||
updated_at = Column(DateTime, default=now_hkt, onupdate=now_hkt)
|
||||
deleted_at = Column(DateTime, nullable=True) # Soft delete
|
||||
|
||||
|
||||
class AttendanceCustomField(Base):
|
||||
__tablename__ = "attendance_custom_fields"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
field_name = Column(String(100), nullable=False)
|
||||
field_type = Column(String(50), nullable=False) # text, number, date, select
|
||||
field_key = Column(String(100), unique=True, nullable=False)
|
||||
required = Column(Boolean, default=False)
|
||||
options = Column(JSON, default=[]) # For select type
|
||||
order = Column(Integer, default=0)
|
||||
created_at = Column(DateTime, default=now_hkt)
|
||||
|
||||
|
||||
class AccidentCustomField(Base):
|
||||
__tablename__ = "accident_custom_fields"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
field_name = Column(String(100), nullable=False)
|
||||
field_type = Column(String(50), nullable=False) # text, number, date, select
|
||||
field_key = Column(String(100), unique=True, nullable=False)
|
||||
required = Column(Boolean, default=False)
|
||||
options = Column(JSON, default=[]) # For select type
|
||||
order = Column(Integer, default=0)
|
||||
created_at = Column(DateTime, default=now_hkt)
|
||||
@@ -0,0 +1,11 @@
|
||||
fastapi==0.115.0
|
||||
uvicorn[standard]==0.30.6
|
||||
sqlalchemy==2.0.35
|
||||
pydantic==2.9.2
|
||||
python-jose[cryptography]==3.3.0
|
||||
passlib==1.7.4
|
||||
python-multipart==0.0.12
|
||||
openpyxl==3.1.5
|
||||
reportlab==4.2.5
|
||||
xlsxwriter==3.2.0
|
||||
aiofiles==24.1.0
|
||||
@@ -0,0 +1,187 @@
|
||||
from pydantic import BaseModel, EmailStr, Field
|
||||
from typing import Optional, List, Dict, Any
|
||||
from datetime import date, time, datetime
|
||||
|
||||
|
||||
# ============ Auth ============
|
||||
class LoginRequest(BaseModel):
|
||||
email: str
|
||||
password: str
|
||||
|
||||
|
||||
class TokenResponse(BaseModel):
|
||||
access_token: str
|
||||
token_type: str = "bearer"
|
||||
|
||||
|
||||
class UserResponse(BaseModel):
|
||||
id: int
|
||||
email: str
|
||||
name: str
|
||||
role: str
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ============ Attendance ============
|
||||
class AttendanceBase(BaseModel):
|
||||
employee_id: str
|
||||
employee_name: str
|
||||
department: Optional[str] = None
|
||||
date: date
|
||||
check_in: Optional[time] = None
|
||||
check_out: Optional[time] = None
|
||||
overtime_hours: float = 0
|
||||
leave_type: Optional[str] = None
|
||||
leave_hours: float = 0
|
||||
status: str
|
||||
remarks: Optional[str] = None
|
||||
custom_fields: Dict[str, Any] = {}
|
||||
|
||||
|
||||
class AttendanceCreate(AttendanceBase):
|
||||
pass
|
||||
|
||||
|
||||
class AttendanceUpdate(BaseModel):
|
||||
employee_id: Optional[str] = None
|
||||
employee_name: Optional[str] = None
|
||||
department: Optional[str] = None
|
||||
date: Optional[date] = None
|
||||
check_in: Optional[time] = None
|
||||
check_out: Optional[time] = None
|
||||
overtime_hours: Optional[float] = None
|
||||
leave_type: Optional[str] = None
|
||||
leave_hours: Optional[float] = None
|
||||
status: Optional[str] = None
|
||||
remarks: Optional[str] = None
|
||||
custom_fields: Optional[Dict[str, Any]] = None
|
||||
|
||||
|
||||
class AttendanceResponse(AttendanceBase):
|
||||
id: int
|
||||
created_by: Optional[int] = None
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ============ Accident ============
|
||||
class AccidentBase(BaseModel):
|
||||
date: date
|
||||
time: Optional[time] = None
|
||||
location: str
|
||||
employee_name: str
|
||||
employee_id: Optional[str] = None
|
||||
department: Optional[str] = None
|
||||
description: str
|
||||
severity: str
|
||||
medical_report: Optional[str] = None
|
||||
action_taken: Optional[str] = None
|
||||
responsible_person: Optional[str] = None
|
||||
custom_fields: Dict[str, Any] = {}
|
||||
|
||||
|
||||
class AccidentCreate(AccidentBase):
|
||||
pass
|
||||
|
||||
|
||||
class AccidentUpdate(BaseModel):
|
||||
date: Optional[date] = None
|
||||
time: Optional[time] = None
|
||||
location: Optional[str] = None
|
||||
employee_name: Optional[str] = None
|
||||
employee_id: Optional[str] = None
|
||||
department: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
severity: Optional[str] = None
|
||||
medical_report: Optional[str] = None
|
||||
action_taken: Optional[str] = None
|
||||
responsible_person: Optional[str] = None
|
||||
custom_fields: Optional[Dict[str, Any]] = None
|
||||
|
||||
|
||||
class AccidentResponse(AccidentBase):
|
||||
id: int
|
||||
created_by: Optional[int] = None
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ============ Custom Fields ============
|
||||
class CustomFieldBase(BaseModel):
|
||||
field_name: str
|
||||
field_type: str
|
||||
field_key: str
|
||||
required: bool = False
|
||||
options: List[str] = []
|
||||
order: int = 0
|
||||
|
||||
|
||||
class CustomFieldCreate(CustomFieldBase):
|
||||
pass
|
||||
|
||||
|
||||
class CustomFieldResponse(CustomFieldBase):
|
||||
id: int
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ============ Dashboard ============
|
||||
class AttendanceStats(BaseModel):
|
||||
total: int
|
||||
today: int
|
||||
this_month: int
|
||||
by_status: Dict[str, int]
|
||||
by_department: Dict[str, int]
|
||||
trend: List[Dict[str, Any]]
|
||||
|
||||
|
||||
class AccidentStats(BaseModel):
|
||||
total: int
|
||||
this_month: int
|
||||
unresolved: int
|
||||
by_severity: Dict[str, int]
|
||||
recent: List[AccidentResponse]
|
||||
|
||||
|
||||
# ============ Import ============
|
||||
class ImportRequest(BaseModel):
|
||||
section: str # "attendance" or "accident"
|
||||
update_existing: bool = True
|
||||
|
||||
|
||||
class ImportResult(BaseModel):
|
||||
created: int
|
||||
updated: int
|
||||
skipped: int
|
||||
errors: List[str]
|
||||
|
||||
|
||||
# ============ Pagination / Filter ============
|
||||
class FilterParams(BaseModel):
|
||||
page: int = 1
|
||||
per_page: int = 20
|
||||
sort_by: Optional[str] = None
|
||||
sort_order: Optional[str] = "asc" # asc or desc
|
||||
search: Optional[str] = None
|
||||
date_from: Optional[date] = None
|
||||
date_to: Optional[date] = None
|
||||
department: Optional[str] = None
|
||||
status: Optional[str] = None
|
||||
|
||||
|
||||
class PaginatedResponse(BaseModel):
|
||||
items: List[Any]
|
||||
total: int
|
||||
page: int
|
||||
per_page: int
|
||||
total_pages: int
|
||||
@@ -0,0 +1,2 @@
|
||||
# AARS JWT Secret - CHANGE THIS IN PRODUCTION!
|
||||
JWT_SECRET = "aars-jwt-secret-2026-prod-do-not-share"
|
||||
@@ -0,0 +1,65 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Quick test script for AARS backend - run locally without Docker"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
# Add backend to path
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
# Generate secret if missing
|
||||
if not os.path.exists("secret.py"):
|
||||
with open("secret.py", "w") as f:
|
||||
f.write(f"JWT_SECRET = 'dev-secret-{os.urandom(16).hex()}'\n")
|
||||
|
||||
# Import and run
|
||||
from database import init_db, SessionLocal
|
||||
from models import User
|
||||
|
||||
print("🧪 Testing AARS Backend...")
|
||||
print()
|
||||
|
||||
# Test 1: Database init
|
||||
print("1️⃣ Testing database initialization...")
|
||||
try:
|
||||
init_db()
|
||||
print(" ✅ Database initialized")
|
||||
except Exception as e:
|
||||
print(f" ❌ Database init failed: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
# Test 2: Check admin user
|
||||
print("2️⃣ Testing admin user...")
|
||||
try:
|
||||
db = SessionLocal()
|
||||
admin = db.query(User).filter(User.email == "admin@aars.hk").first()
|
||||
if admin:
|
||||
print(f" ✅ Admin exists: {admin.email}")
|
||||
else:
|
||||
print(" ⚠️ Admin not found")
|
||||
db.close()
|
||||
except Exception as e:
|
||||
print(f" ❌ Query failed: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
# Test 3: Test password hashing
|
||||
print("3️⃣ Testing password hashing...")
|
||||
try:
|
||||
from database import get_password_hash, verify_password
|
||||
pw_hash = get_password_hash("test123")
|
||||
assert verify_password("test123", pw_hash)
|
||||
print(" ✅ Password hashing works")
|
||||
except Exception as e:
|
||||
print(f" ❌ Password hashing failed: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
print()
|
||||
print("🎉 All basic tests passed!")
|
||||
print()
|
||||
print("To run the backend:")
|
||||
print(" cd backend")
|
||||
print(" pip install -r requirements.txt")
|
||||
print(" uvicorn main:app --reload --port 8000")
|
||||
print()
|
||||
print("Then visit: http://localhost:8000")
|
||||
print("API docs: http://localhost:8000/docs")
|
||||
Reference in New Issue
Block a user