initial commit

.gitignore

@@ -0,0 +1,12 @@
+__pycache__/
+*.py[cod]
+*.pyo
+*.pyd
+*.db
+*.sqlite3
+*.log
+*.env
+*.venv
+*.egg-info/
+dist/
+build/

Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12.7
+WORKDIR /code
+COPY ./requirements.txt /code/requirements.txt
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+RUN useradd user
+USER user
+
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH
+
+WORKDIR $HOME/app
+
+COPY --chown=user . $HOME/app
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

app/auth.py

@@ -0,0 +1,57 @@
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from sqlalchemy.orm import Session
+
+from . import crud, models, schemas
+from .database import get_db
+from .settings import settings
+
+# Password Hashing
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+# OAuth2 Scheme
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
+
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.now(timezone.utc) + expires_delta
+    else:
+        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
+    return encoded_jwt
+
+def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
+        username: str = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = schemas.TokenData(username=username)
+    except JWTError:
+        raise credentials_exception
+    user = crud.get_user_by_username(db, username=token_data.username)
+    if user is None:
+        raise credentials_exception
+    return user
+
+def get_current_active_teacher(current_user: models.User = Depends(get_current_user)):
+    if current_user.role != models.UserRole.teacher:
+        raise HTTPException(status_code=403, detail="Not authorized")
+    return current_user

app/crud.py

@@ -0,0 +1,41 @@
+from sqlalchemy.orm import Session
+from . import models, schemas, auth
+
+def get_user_by_username(db: Session, username: str):
+    return db.query(models.User).filter(models.User.username == username).first()
+
+def create_user(db: Session, user: schemas.UserCreate):
+    hashed_password = auth.get_password_hash(user.password)
+    db_user = models.User(
+        username=user.username,
+        hashed_password=hashed_password,
+        full_name=user.full_name,
+        role=user.role
+    )
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
+
+def get_lectures_by_day(db: Session, day_of_week: int, teacher_id: int):
+    return db.query(models.TimetableSlot).filter(
+        models.TimetableSlot.day_of_week == day_of_week,
+        models.TimetableSlot.teacher_id == teacher_id
+    ).all()
+
+def check_if_already_marked(db: Session, student_id: int, session_id: str):
+    return db.query(models.AttendanceRecord).filter(
+        models.AttendanceRecord.student_id == student_id,
+        models.AttendanceRecord.session_id == session_id
+    ).first() is not None
+
+def create_attendance_record(db: Session, student_id: int, slot_id: int, session_id: str):
+    db_record = models.AttendanceRecord(
+        student_id=student_id,
+        slot_id=slot_id,
+        session_id=session_id
+    )
+    db.add(db_record)
+    db.commit()
+    db.refresh(db_record)
+    return db_record

app/database.py

@@ -0,0 +1,26 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+import redis
+from .settings import settings
+
+# PostgreSQL Setup
+engine = create_engine(settings.database_url)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+Base = declarative_base()
+
+# Redis Setup
+redis_client = redis.from_url(settings.redis_url, decode_responses=True)
+
+
+# Dependency to get a DB session
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+# Dependency to get a Redis client
+def get_redis():
+    return redis_client

app/models.py

@@ -0,0 +1,46 @@
+from sqlalchemy import Column, Integer, String, ForeignKey, Time, DateTime, Enum as SQLAlchemyEnum
+from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
+import enum
+from .database import Base
+
+class UserRole(enum.Enum):
+    teacher = "teacher"
+    student = "student"
+
+class User(Base):
+    __tablename__ = "users"
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String, unique=True, index=True, nullable=False)
+    hashed_password = Column(String, nullable=False)
+    full_name = Column(String, nullable=False)
+    role = Column(SQLAlchemyEnum(UserRole), nullable=False)
+
+class Subject(Base):
+    __tablename__ = "subjects"
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String, nullable=False)
+    code = Column(String, unique=True, nullable=False)
+
+class TimetableSlot(Base):
+    __tablename__ = "timetable_slots"
+    id = Column(Integer, primary_key=True, index=True)
+    subject_id = Column(Integer, ForeignKey("subjects.id"), nullable=False)
+    teacher_id = Column(Integer, ForeignKey("users.id"), nullable=False)
+    day_of_week = Column(Integer, nullable=False) # 0=Monday, 1=Tuesday...
+    start_time = Column(Time, nullable=False)
+    end_time = Column(Time, nullable=False)
+
+    subject = relationship("Subject")
+    teacher = relationship("User")
+
+class AttendanceRecord(Base):
+    __tablename__ = "attendance_records"
+    id = Column(Integer, primary_key=True, index=True)
+    student_id = Column(Integer, ForeignKey("users.id"), nullable=False)
+    slot_id = Column(Integer, ForeignKey("timetable_slots.id"), nullable=False)
+    session_id = Column(String, nullable=False, index=True)
+    timestamp = Column(DateTime(timezone=True), server_default=func.now())
+    
+    student = relationship("User")
+    slot = relationship("TimetableSlot")

app/routers/auth.py

@@ -0,0 +1,39 @@
+from datetime import timedelta
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy.orm import Session
+
+from .. import auth, crud, schemas, models
+from ..database import get_db
+from ..settings import settings
+
+router = APIRouter(
+    prefix="/auth",
+    tags=["Authentication"],
+)
+
+@router.post("/token", response_model=schemas.Token)
+def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
+    user = crud.get_user_by_username(db, username=form_data.username)
+    if not user or not auth.verify_password(form_data.password, user.hashed_password):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(minutes=settings.access_token_expire_minutes)
+    access_token = auth.create_access_token(
+        data={"sub": user.username}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+@router.get("/me", response_model=schemas.User)
+def read_users_me(current_user: models.User = Depends(auth.get_current_user)):
+    """
+    Get the current logged-in user's details.
+    
+    This endpoint is protected. It uses the `get_current_user` dependency 
+    to validate the JWT token from the Authorization header and return 
+    the corresponding user's data from the database.
+    """
+    return current_user

app/routers/student.py

@@ -0,0 +1,56 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from redis import Redis
+from sqlalchemy.orm import Session
+import json
+
+from .. import auth, crud, models, schemas
+from ..database import get_db, get_redis
+from .teacher import manager # Import the connection manager
+
+router = APIRouter(
+    prefix="/student",
+    tags=["Student"],
+    dependencies=[Depends(auth.get_current_user)]
+)
+
+@router.post("/attendance/scan")
+async def scan_attendance(
+    payload: schemas.AttendanceScanPayload,
+    db: Session = Depends(get_db),
+    redis_client: Redis = Depends(get_redis),
+    current_user: models.User = Depends(auth.get_current_user)
+):
+    if current_user.role != models.UserRole.student:
+        raise HTTPException(status_code=403, detail="Only students can scan for attendance")
+    
+    # 1. Verify Token
+    valid_token = redis_client.get(f"qr_token:{payload.session_id}")
+    if not valid_token or valid_token != payload.token:
+        raise HTTPException(status_code=400, detail="Invalid or expired QR code")
+    
+    # 2. Get lecture ID for this session
+    slot_id = redis_client.get(f"session_slot:{payload.session_id}")
+    if not slot_id:
+        raise HTTPException(status_code=404, detail="Attendance session not found")
+
+    # 3. Check for duplicates
+    if crud.check_if_already_marked(db, student_id=current_user.id, session_id=payload.session_id):
+        raise HTTPException(status_code=409, detail="Attendance already marked for this session")
+
+    # 4. Mark attendance
+    record = crud.create_attendance_record(
+        db,
+        student_id=current_user.id,
+        slot_id=int(slot_id),
+        session_id=payload.session_id
+    )
+
+    # 5. Notify teacher via WebSocket
+    student_data = {
+        "id": current_user.id,
+        "name": current_user.full_name,
+        "email": current_user.username
+    }
+    await manager.broadcast_to_session(json.dumps(student_data), payload.session_id)
+
+    return {"status": "success", "message": "Attendance marked successfully"}

app/routers/teacher.py

@@ -0,0 +1,180 @@
+# import uuid
+# import asyncio
+# from datetime import date
+# from typing import List, Dict
+
+# from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect, HTTPException
+# from redis import Redis
+# from sqlalchemy.orm import Session
+
+# from .. import auth, crud, models, schemas
+# from ..database import get_db, get_redis
+
+# router = APIRouter(
+#     prefix="/teacher",
+#     tags=["Teacher"],
+#     dependencies=[Depends(auth.get_current_active_teacher)]
+# )
+
+# class ConnectionManager:
+#     def __init__(self):
+#         self.active_connections: Dict[str, WebSocket] = {}
+
+#     async def connect(self, websocket: WebSocket, session_id: str):
+#         await websocket.accept()
+#         self.active_connections[session_id] = websocket
+
+#     def disconnect(self, session_id: str):
+#         if session_id in self.active_connections:
+#             del self.active_connections[session_id]
+
+#     async def send_personal_message(self, message: str, websocket: WebSocket):
+#         await websocket.send_text(message)
+    
+#     async def broadcast_to_session(self, message: str, session_id: str):
+#         if session_id in self.active_connections:
+#             await self.active_connections[session_id].send_text(message)
+
+# manager = ConnectionManager()
+
+# @router.get("/lectures", response_model=List[schemas.TimetableSlot])
+# def get_teacher_lectures_for_date(
+#     selected_date: date,
+#     db: Session = Depends(get_db),
+#     current_user: models.User = Depends(auth.get_current_active_teacher)
+# ):
+#     day_of_week = selected_date.weekday()
+#     return crud.get_lectures_by_day(db, day_of_week=day_of_week, teacher_id=current_user.id)
+
+# @router.post("/attendance/start")
+# def start_attendance_session(
+#     slot_id: int,
+#     redis_client: Redis = Depends(get_redis)
+# ):
+#     session_id = str(uuid.uuid4())
+#     token = str(uuid.uuid4())
+    
+#     # Store token and associated lecture (slot_id) in Redis
+#     redis_client.set(f"qr_token:{session_id}", token, ex=20) # ex=20 -> expires in 20 seconds
+#     redis_client.set(f"session_slot:{session_id}", slot_id)
+    
+#     return {"session_id": session_id, "initial_token": token}
+
+# @router.get("/attendance/qr/{session_id}")
+# def refresh_qr_token(session_id: str, redis_client: Redis = Depends(get_redis)):
+#     if not redis_client.exists(f"session_slot:{session_id}"):
+#         raise HTTPException(status_code=404, detail="Attendance session not found or expired")
+
+#     new_token = str(uuid.uuid4())
+#     redis_client.set(f"qr_token:{session_id}", new_token, ex=20)
+#     return {"token": new_token}
+
+# @router.websocket("/ws/attendance/{session_id}")
+# async def websocket_endpoint(websocket: WebSocket, session_id: str):
+#     await manager.connect(websocket, session_id)
+#     try:
+#         while True:
+#             # Keep connection alive
+#             await asyncio.sleep(1)
+#     except WebSocketDisconnect:
+#         manager.disconnect(session_id)
+
+import uuid
+import asyncio
+from datetime import date
+from typing import List, Dict
+
+from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect, HTTPException, Query, status
+from redis import Redis
+from sqlalchemy.orm import Session
+
+from .. import auth, crud, models, schemas
+from ..database import get_db, get_redis
+
+# --- NEW: WebSocket-specific authenticator ---
+# This function will be used ONLY by the websocket endpoint.
+# It looks for the token in a query parameter instead of a header.
+async def get_current_user_from_query(
+    token: str = Query(...), 
+    db: Session = Depends(get_db)
+):
+    user = auth.get_current_user(token=token, db=db)
+    if user.role != models.UserRole.teacher:
+        raise WebSocketDisconnect(code=status.WS_1008_POLICY_VIOLATION, reason="Not authorized")
+    return user
+
+
+# --- MODIFIED: Router definition ---
+# We REMOVE the global dependency from the router itself.
+router = APIRouter(
+    prefix="/teacher",
+    tags=["Teacher"]
+)
+
+class ConnectionManager:
+    def __init__(self):
+        self.active_connections: Dict[str, WebSocket] = {}
+
+    async def connect(self, websocket: WebSocket, session_id: str):
+        await websocket.accept()
+        self.active_connections[session_id] = websocket
+
+    def disconnect(self, session_id: str):
+        if session_id in self.active_connections:
+            del self.active_connections[session_id]
+    
+    async def broadcast_to_session(self, message: str, session_id: str):
+        if session_id in self.active_connections:
+            await self.active_connections[session_id].send_text(message)
+
+manager = ConnectionManager()
+
+# --- MODIFIED: Add the dependency directly to the HTTP routes ---
+@router.get("/lectures", response_model=List[schemas.TimetableSlot], dependencies=[Depends(auth.get_current_active_teacher)])
+def get_teacher_lectures_for_date(
+    selected_date: date,
+    db: Session = Depends(get_db),
+    current_user: models.User = Depends(auth.get_current_active_teacher)
+):
+    day_of_week = selected_date.weekday()
+    return crud.get_lectures_by_day(db, day_of_week=day_of_week, teacher_id=current_user.id)
+
+@router.post("/attendance/start", dependencies=[Depends(auth.get_current_active_teacher)])
+def start_attendance_session(
+    slot_id: int,
+    redis_client: Redis = Depends(get_redis)
+):
+    session_id = str(uuid.uuid4())
+    token = str(uuid.uuid4())
+    
+    redis_client.set(f"qr_token:{session_id}", token, ex=20)
+    redis_client.set(f"session_slot:{session_id}", slot_id)
+    
+    return {"session_id": session_id, "initial_token": token}
+
+@router.get("/attendance/qr/{session_id}", dependencies=[Depends(auth.get_current_active_teacher)])
+def refresh_qr_token(session_id: str, redis_client: Redis = Depends(get_redis)):
+    if not redis_client.exists(f"session_slot:{session_id}"):
+        raise HTTPException(status_code=404, detail="Attendance session not found or expired")
+
+    new_token = str(uuid.uuid4())
+    redis_client.set(f"qr_token:{session_id}", new_token, ex=20)
+    print( f"Refreshed token for session {session_id}: {new_token}" )
+    return {"token": new_token}
+
+# --- MODIFIED: WebSocket endpoint now uses its own authenticator ---
+@router.websocket("/ws/attendance/{session_id}")
+async def websocket_endpoint(
+    websocket: WebSocket,
+    session_id: str,
+    # This dependency will now be resolved from the query parameter
+    user: models.User = Depends(get_current_user_from_query) 
+):
+    await manager.connect(websocket, session_id)
+    try:
+        while True:
+            # Keep connection alive
+            await asyncio.sleep(1)
+    except WebSocketDisconnect:
+        manager.disconnect(session_id)
+

app/schemas.py

@@ -0,0 +1,56 @@
+from pydantic import BaseModel
+from typing import List
+import datetime
+from .models import UserRole
+
+# --- User Schemas ---
+class UserBase(BaseModel):
+    username: str
+    full_name: str
+
+class UserCreate(UserBase):
+    password: str
+    role: UserRole
+
+class User(UserBase):
+    id: int
+    role: UserRole
+
+    class Config:
+        from_attributes = True
+
+# --- Auth Schemas ---
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+class TokenData(BaseModel):
+    username: str | None = None
+
+# --- Timetable Schemas ---
+class Subject(BaseModel):
+    id: int
+    name: str
+    code: str
+
+    class Config:
+        from_attributes = True
+
+class TimetableSlot(BaseModel):
+    id: int
+    day_of_week: int
+    start_time: datetime.time
+    end_time: datetime.time
+    subject: Subject
+    
+    class Config:
+        from_attributes = True
+
+# --- Attendance Schemas ---
+class AttendanceScanPayload(BaseModel):
+    token: str
+    session_id: str
+
+class LiveAttendanceUpdate(BaseModel):
+    student_name: str
+    timestamp: datetime.datetime

app/settings.py

@@ -0,0 +1,14 @@
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    """Loads environment variables from the .env file."""
+    database_url: str
+    redis_url: str
+    secret_key: str
+    algorithm: str
+    access_token_expire_minutes: int
+
+    class Config:
+        env_file = ".\.env"
+
+settings = Settings()

main.py

@@ -0,0 +1,35 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from app import models
+from app.database import engine
+from app.routers import auth, teacher, student
+
+# Create all database tables
+models.Base.metadata.create_all(bind=engine)
+
+app = FastAPI()
+
+# CORS Middleware Setup
+# Replace "http://localhost:3000" with your Next.js frontend URL
+origins = [
+    "http://localhost:3000",
+    "http://127.0.0.1:63442",
+    "http://localhost:9002",
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Include Routers
+app.include_router(auth.router)
+app.include_router(teacher.router)
+app.include_router(student.router)
+
+@app.get("/")
+def read_root():
+    return {"message": "Welcome to the Smart Attendance System API"}

requirements.txt

@@ -0,0 +1,11 @@
+fastapi==0.111.0
+uvicorn[standard]==0.29.0
+sqlalchemy==2.0.30
+psycopg2-binary==2.9.9
+redis==5.0.4
+pydantic==2.7.1
+pydantic-settings==2.2.1
+python-jose[cryptography]==3.3.0
+passlib[bcrypt]
+python-multipart==0.0.9
+websockets==12.0