import re import io from typing import Optional from fastapi import APIRouter, Depends, Form, Query, UploadFile, File, HTTPException, status from sqlalchemy import distinct, func, case, or_ from sqlalchemy.orm import Session from fastapi.responses import JSONResponse from fastapi.encoders import jsonable_encoder from fastapi.security import OAuth2PasswordRequestForm,HTTPBearer, HTTPAuthorizationCredentials from core.security import verify_password, get_password_hash, create_access_token, verify_token from datetime import datetime, timezone from core.config import settings from db.session import get_db from db.models.user import User # Replace with actual import from db.schemas.user import Token, LoginRequest, UserResponse, UserCreate # Replace with actual import from jose import jwt, JWTError from fastapi import File, UploadFile import os from db.models.user_course_log import UserCourseLog from db.models.course import Course import pandas as pd import numpy as np from db.models.role import Role from db.models.university import University router = APIRouter() security = HTTPBearer() EMAIL_REGEX = r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$' @router.get("/", response_model=UserResponse) async def get_active_users( pageNo: int = 1, recordsPerPage: int = 10, search: Optional[str] = None, sort_by: str = "id", sort_order: str = "asc", role_id: Optional[int] = Query(None), user_status: str = "all", platform_id: int = Query(..., description="Platform ID is required"), db: Session = Depends(get_db), credentials: HTTPAuthorizationCredentials = Depends(security) ): """ Retrieve a paginated list of users filtered by platform_id. Each user's total credited/debited courses are aggregated individually. """ # ✅ Verify token token = credentials.credentials user_id = verify_token(token) user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid user token") # ✅ Validate platform_id try: platform_id = int(platform_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid platform_id format") # ✅ Base query for users db_query = db.query(User).filter(User.platform_id == platform_id) # ✅ Apply user_status if user_status.lower() == "active": db_query = db_query.filter(User.is_active == True) elif user_status.lower() == "inactive": db_query = db_query.filter(User.is_active == False) # ✅ Apply role filter if provided if role_id is not None: db_query = db_query.filter(User.role_id == role_id) # ✅ Apply search filter if search: db_query = db_query.filter( or_( User.name.ilike(f"%{search}%"), User.email.ilike(f"%{search}%") ) ) # ✅ Sorting sort_column = getattr(User, sort_by, None) if not sort_column: raise HTTPException(status_code=400, detail=f"Invalid sort field '{sort_by}'") db_query = db_query.order_by(sort_column.asc() if sort_order == "asc" else sort_column.desc()) # ✅ Pagination offset = (pageNo - 1) * recordsPerPage paginated_users = db_query.offset(offset).limit(recordsPerPage).all() if not paginated_users: raise HTTPException(status_code=404, detail="No users found for this platform") # ✅ Get individual user course totals user_totals = ( db.query( UserCourseLog.user_id, func.sum(case((UserCourseLog.type == "Credit", UserCourseLog.number_of_course), else_=0)).label("total_credited_courses"), func.sum(case((UserCourseLog.type == "Debit", UserCourseLog.number_of_course), else_=0)).label("total_debited_courses") ) .group_by(UserCourseLog.user_id) .all() ) # Convert to dict for easy lookup user_totals_dict = {user_id: {"total_credited_courses": total_credited or 0, "total_debited_courses": total_debited or 0} for user_id, total_credited, total_debited in user_totals} # ✅ Prepare response list users_list = [] for u in paginated_users: totals = user_totals_dict.get(u.id, {"total_credited_courses": 0, "total_debited_courses": 0}) users_list.append({ "id": u.id, "name": u.name, "email": u.email, "role": u.role, "platform_id": u.platform_id, "university_id": u.university_id, "university": { "id": u.university.id, "name": u.university.name, "institution_code": u.university.institution_code } if u.university else None, "activation_date": u.activation_date, "phone_number": u.phone_number, "department": u.department, "is_active": u.is_active, "total_credited_courses": totals["total_credited_courses"], "total_debited_courses": totals["total_debited_courses"] }) # ✅ Total users count total_query = db.query(func.count(User.id)).filter(User.platform_id == platform_id) if user_status.lower() == "active": total_query = total_query.filter(User.is_active == True) elif user_status.lower() == "inactive": total_query = total_query.filter(User.is_active == False) if role_id is not None: total_query = total_query.filter(User.role_id == role_id) total_users = total_query.scalar() or 0 # ✅ Response with pagination response_data = jsonable_encoder({ "users": users_list, "total_count": total_users, "page_no": pageNo, "records_per_page": recordsPerPage, "total_pages": (total_users + recordsPerPage - 1) // recordsPerPage, "status_code": status.HTTP_200_OK, "message": "Users retrieved successfully" }) return JSONResponse(content=response_data, status_code=status.HTTP_200_OK) @router.post("/create", response_model=UserCreate) async def create_user( user: UserCreate, db: Session = Depends(get_db), credentials: HTTPAuthorizationCredentials = Depends(security) ): """ Create a new user. """ token = credentials.credentials user_id = verify_token(token) current_user = db.query(User).filter(User.id == user_id).first() if not current_user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid user token") try: required_fields = [ 'name', 'email', 'password', 'role_id','university_id', 'phone_number', 'activation_date', 'educational_degree', 'department' ] user_data = user.dict() user_data['created_at'] = datetime.now(timezone.utc) user_data['updated_at'] = datetime.now(timezone.utc) user_data['password_hash'] = get_password_hash(user.password), if not re.match(EMAIL_REGEX, user_data['email']): return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={ "status": status.HTTP_400_BAD_REQUEST, "message": "Email: Invalid value." } ) for field in required_fields: if field not in user_data or not user_data[field]: filed_name = field.replace('_', ' ').capitalize() return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={ "status": status.HTTP_400_BAD_REQUEST, "message": f"{filed_name} is required." } ) del user_data['password'] new_user = User(**user_data) db.add(new_user) db.commit() db.refresh(new_user) return JSONResponse( content={ 'user': jsonable_encoder(UserResponse.from_orm(new_user)), 'status_code': status.HTTP_201_CREATED, 'message': 'User created successfully' }, status_code=status.HTTP_201_CREATED ) except Exception as e: raise HTTPException(status_code=500, detail=f"Something went wrong: {str(e)}") @router.get("/view/{user_id}", response_model=UserResponse) async def get_user_by_id( user_id: int, db: Session = Depends(get_db), credentials: HTTPAuthorizationCredentials = Depends(security) ): """ Retrieve a user by ID. """ token = credentials.credentials current_user_id = verify_token(token) current_user = db.query(User).filter(User.id == current_user_id).first() if not current_user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid user token") user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") return JSONResponse( content={ 'user': jsonable_encoder(UserResponse.from_orm(user)), 'status_code': status.HTTP_200_OK, 'message': 'User retrieved successfully' }, status_code=status.HTTP_200_OK ) @router.put("/change-status/{user_id}/{user_status}", response_model=UserResponse) async def change_user_status( user_id: int, user_status: str, db: Session = Depends(get_db), credentials: HTTPAuthorizationCredentials = Depends(security) ): token = credentials.credentials current_user_id = verify_token(token) current_user = db.query(User).filter(User.id == current_user_id).first() if not current_user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid user token") user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") if user_status == "active": user.is_active = True user.activation_date = datetime.now(timezone.utc) elif user_status == "inactive": user.is_active = False user.activation_date = None else: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid status") db.commit() return JSONResponse( content={ 'status_code': status.HTTP_200_OK, 'message': 'User status updated successfully' }, status_code=status.HTTP_200_OK ) @router.put("/update/{user_id}", response_model=UserResponse) async def update_user( user_id: int, user: UserCreate, db: Session = Depends(get_db), credentials: HTTPAuthorizationCredentials = Depends(security) ): """ Update a user's details. """ token = credentials.credentials current_user_id = verify_token(token) current_user = db.query(User).filter(User.id == current_user_id).first() if not current_user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid user token") existing_user = db.query(User).filter(User.id == user_id).first() if not existing_user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") try: existing_user.name = user.name existing_user.email = user.email if user.password: if verify_password(user.password, existing_user.password_hash): # raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Password cannot be the same as the current password") return JSONResponse( content={ 'status_code': status.HTTP_400_BAD_REQUEST, 'message': 'Password cannot be the same as the current password' }, status_code=status.HTTP_400_BAD_REQUEST ) existing_user.password_hash = get_password_hash(user.password) if user.password else existing_user.password_hash existing_user.is_active = user.is_active existing_user.role_id = user.role_id existing_user.activation_date = user.activation_date if user.activation_date else None existing_user.is_email_verified = user.is_email_verified if user.is_email_verified else False existing_user.is_phone_verified = user.is_phone_verified if user.is_phone_verified else False existing_user.phone_number = user.phone_number if user.phone_number else None existing_user.profile_picture = user.profile_picture if user.profile_picture else None existing_user.department = user.department if user.department else None existing_user.educational_degree = user.educational_degree if user.educational_degree else None existing_user.subscription_package = user.subscription_package if user.subscription_package else None existing_user.subscription_start_date = user.subscription_start_date if user.subscription_start_date else None existing_user.subscription_end_date = user.subscription_end_date if user.subscription_end_date else None existing_user.subscription_status = user.subscription_status if user.subscription_status else None existing_user.subscription_renewal_date = user.subscription_renewal_date if user.subscription_renewal_date else None existing_user.updated_at = datetime.now(timezone.utc) # Update the timestamp db.commit() db.refresh(existing_user) return JSONResponse( content={ 'user': jsonable_encoder(UserResponse.from_orm(existing_user)), 'status_code': status.HTTP_200_OK, 'message': 'User updated successfully' }, status_code=status.HTTP_200_OK ) except Exception as e: raise HTTPException(status_code=500, detail=f"Something went wrong: {str(e)}") @router.delete("/delete/{user_id}", response_model=UserResponse) async def delete_user( user_id: int, db: Session = Depends(get_db), credentials: HTTPAuthorizationCredentials = Depends(security) ): """ Delete a user by ID. """ token = credentials.credentials current_user_id = verify_token(token) current_user = db.query(User).filter(User.id == current_user_id).first() if not current_user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid user token") user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") db.delete(user) db.commit() return JSONResponse( content={ 'status_code': status.HTTP_200_OK, 'message': 'User deleted successfully' }, status_code=status.HTTP_200_OK ) @router.post("/upload/") async def upload_file( file: UploadFile = File(...), type: str = Form(...) ): """ Uploads a single file and saves it to the local directory with timestamp. """ MAX_FILE_SIZE = 2 * 1024 * 1024 # 2 MB ALLOWED_EXTENSIONS = ["jpg", "jpeg", "png", "svg"] try: # Validate type if type not in ["profile", "logo"]: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid file type") # Validate file extension ext = file.filename.split('.')[-1].lower() if ext not in ALLOWED_EXTENSIONS: return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={ "status":status.HTTP_400_BAD_REQUEST, "message": f"Invalid file extension. Allowed: {', '.join(ALLOWED_EXTENSIONS)}" } ) # Read file content contents = await file.read() # Validate file size if len(contents) > MAX_FILE_SIZE: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="File size exceeds 2MB limit.") # Ensure directories exist upload_dir = f"uploads/{type}" os.makedirs(upload_dir, exist_ok=True) # Read file content # contents = await file.read() # print(f"Uploaded file: {file.filename}") # Add timestamp to filename timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") new_filename = f"{timestamp}_{file.filename}" file_path = os.path.join(upload_dir, new_filename) # Save file with open(file_path, "wb") as f: f.write(contents) return JSONResponse( {"filename": 'uploads/'+type+"/"+new_filename, "message": "File uploaded successfully ✅"}, status_code=status.HTTP_200_OK ) except Exception as e: return JSONResponse({"error": str(e)}, status_code=500) UPLOAD_DIR = "uploads/excel" os.makedirs(UPLOAD_DIR, exist_ok=True) EMAIL_REGEX = r"^[\w\.-]+@[\w\.-]+\.\w+$" @router.post("/upload-excel") async def upload_excel( file: UploadFile = File(...), db: Session = Depends(get_db) ): """ Upload Excel file (.xls or .xlsx), validate all rows first. If all rows are valid → insert all records. If any row has error → return all errors, no insertion happens. """ try: # ✅ Validate file extension if not file.filename.lower().endswith((".xls", ".xlsx")): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid file format. Only .xls or .xlsx allowed." ) # ✅ Save temporarily timestamp = datetime.now().strftime("%Y%m%d%H%M%S") file_path = os.path.join(UPLOAD_DIR, f"{timestamp}_{file.filename}") with open(file_path, "wb") as f: f.write(await file.read()) # ✅ Read Excel using pandas try: df = pd.read_excel(file_path) except Exception as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unable to read Excel file: {str(e)}" ) df = df.replace([np.nan, np.inf, -np.inf], None) # ✅ Required fields required_fields = [ 'name', 'email', 'password', 'role', 'institution_code', 'phone_number', 'activation_date', 'educational_degree', 'department', 'subscription_package', 'subscription_start_date', 'subscription_end_date', 'subscription_status', 'subscription_renewal_date' ] # ✅ Check if all required columns exist missing_cols = [col for col in required_fields if col not in df.columns] if missing_cols: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Missing columns in Excel: {', '.join(missing_cols)}" ) errors = [] valid_users = [] # ✅ Validate each row for index, row in df.iterrows(): row_number = index + 2 # Excel header + 1 user_data = {col: row[col] for col in required_fields} row_errors = [] # ✅ Required field validation for field in required_fields: if not user_data[field]: row_errors.append(f"{field.replace('_', ' ').capitalize()} is required.") # ✅ Email validation if user_data.get("email") and not re.match(EMAIL_REGEX, str(user_data["email"])): row_errors.append(f"Invalid email format: {user_data['email']}") # ✅ Role validation role_data = None if user_data.get("role"): role_data = db.query(Role).filter(Role.name == user_data["role"]).first() if not role_data: row_errors.append(f"Invalid role: {user_data['role']}") else: row_errors.append("Role is required.") # ✅ University validation university_data = None if user_data.get("institution_code"): university_data = db.query(University).filter( University.institution_code == user_data["institution_code"] ).first() if not university_data: row_errors.append(f"Invalid university code: {user_data['institution_code']}") else: row_errors.append("Institution code is required.") # ✅ Append row errors or valid data if row_errors: errors.append({ "row": row_number, "errors": row_errors }) else: user_data["role_id"] = role_data.id user_data["university_id"] = university_data.id valid_users.append(user_data) # ❌ If any errors → return them and stop insertion if errors: return JSONResponse( content={ "status": "error", "message": "Validation failed. Please fix the errors before re-uploading.", "total_rows": len(df), "failed_rows": len(errors), "errors": errors }, status_code=status.HTTP_400_BAD_REQUEST, ) # ✅ Insert all valid users for user_data in valid_users: try: new_user = User( name=user_data["name"], email=user_data["email"], password_hash=get_password_hash(str(user_data["password"])), phone_number=user_data["phone_number"], role_id=user_data["role_id"], university_id=user_data["university_id"], activation_date=user_data["activation_date"], educational_degree=user_data["educational_degree"], department=user_data["department"], subscription_package=user_data["subscription_package"], subscription_start_date=user_data["subscription_start_date"], subscription_end_date=user_data["subscription_end_date"], subscription_status=user_data["subscription_status"], subscription_renewal_date=user_data["subscription_renewal_date"], is_active=True, created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc) ) db.add(new_user) except Exception as e: errors.append({ "row": row_number, "errors": [f"Database error: {str(e)}"] }) # ✅ Final commit if errors: db.rollback() return JSONResponse( content={ "status": "error", "message": "Some rows failed to insert.", "errors": errors }, status_code=status.HTTP_400_BAD_REQUEST, ) db.commit() return JSONResponse( content={ "status": "success", "message": f"{len(valid_users)} users imported successfully.", "inserted_users": [u["email"] for u in valid_users], }, status_code=status.HTTP_200_OK, ) except HTTPException as e: raise e except Exception as e: db.rollback() raise HTTPException( status_code=500, detail=f"Internal server error: {str(e)}" )