server.py

from fastapi import FastAPI, Request  , HTTPException
from fastapi.responses import HTMLResponse , RedirectResponse , JSONResponse 
from dotenv import load_dotenv
import os
import threading
import traceback
from logger import log_error, log_info
from db_operations import read_admin_user , read_all_quotes , block_account , Quote , write_quote , read_quotes_by_author , delete_quote_by_id
from password_hashing import verify_password
from jwt_token_operations import create_access_token, read_access_token


SESSION = {}

load_dotenv()

app = FastAPI()

@app.get('/health', response_class=JSONResponse)
async def health_check():
    return JSONResponse(content={"status": "ok"}, status_code=200)

@app.post('/login',response_class=JSONResponse)
async def login(request: Request):
    try:
        request_data = await request.json()
        username = request_data.get("username")
        password = request_data.get("password")
        if not username or not password:
            if username in SESSION:
                del SESSION[username]
            return JSONResponse(content={"message": "Username and password are required"}, status_code=200)
        
        response , status = read_admin_user(username)

        response_data = { 
            "status": False,
            "account_blocked": False, 
            "reason": "", 
            "attempts_left": max(0, 3 - SESSION[username]["attempts"]) if username in SESSION else 3
        }

        if not status:
            response_data["reason"] = "User not found"
            if username in SESSION:
              del SESSION[username]
            return JSONResponse(content=response_data, status_code=200)
        
        if response.blocked:
            response_data['account_blocked'] = True
            response_data['reason'] = "Your account has been blocked due to multiple failed login attempts. Please contact support."
            response_data['attempts_left'] = 0
            if username in SESSION:
                del SESSION[username]
            return JSONResponse(content=response_data, status_code=200)
        
        correct_password = verify_password(password, response.password)

        if not correct_password:
            if username in SESSION:
                SESSION[username]["attempts"] += 1
            else:
                SESSION[username] = {"attempts": 1, "blocked": False}

            if SESSION[username]["attempts"] >= 3:
                blocked = block_account(username)
                if blocked:
                    log_error(f"blocked account for username: {username}")
                    SESSION[username]["blocked"] = True

            response_data["account_blocked"] = SESSION[username]["blocked"] if username in SESSION else False
            response_data["reason"] = "Invalid username or password"
            response_data["attempts_left"] = max(0, 3 - SESSION[username]["attempts"]) if username in SESSION else 3

            if SESSION[username]["blocked"]:
                response_data["reason"] = "Your account has been blocked due to multiple failed login attempts. Please contact support."
        
            return JSONResponse(content=response_data, status_code=200)

        response_data["reason"] = "Login successful"
        response_data["attempts_left"] = SESSION[username]["attempts"] if username in SESSION else 0
        response_data["account_blocked"] = False
        response_data["status"] = True

        if username in SESSION:
            del SESSION[username]

        response = JSONResponse(content=response_data, status_code=200)
        access_token = create_access_token(username=username)
        if access_token:
            response.set_cookie(key="access_token", value=access_token, httponly=True, secure=True, samesite="None",max_age=900)

        return response

    except Exception as e:
        exception_traceback = traceback.format_exc()
        error_name = type(e).__name__
        log_error(message=f"Location : login - file : server.py : Error Name - {error_name}",exc_info=exception_traceback)
        return JSONResponse(content={ 
                "status": False,
                "account_blocked": SESSION[username]["blocked"] if username in SESSION else False, 
                "reason": "Internal server error your account is not blocked but you can try again after sometime", 
                "attempts_left": max(0, 3 - SESSION[username]["attempts"]) if username in SESSION else 0
            }, status_code=200)

@app.get('/auth_check',response_class=JSONResponse)
async def auth_check(request: Request):
    try:
        access_token = request.cookies.get("access_token")

        if not access_token:
            return JSONResponse(content={"logged_in":False},status_code=200)
        token_data = read_access_token(access_token)


        if not token_data.get("success"):
            
            response = JSONResponse(content={"logged_in":False},status_code=200)

            response.delete_cookie("access_token")

            return response


        return JSONResponse(content={"logged_in":True,"username":token_data["username"]},status_code=200)
    
    except Exception as e:
        exception_traceback = traceback.format_exc()
        error_name = type(e).__name__
        log_error(message=f"Location : auth_check - file : server.py : Error Name - {error_name}",exc_info=exception_traceback)
        return JSONResponse(content={"logged_in": False}, status_code=200)
    
@app.get('/quote_fetch',response_class=JSONResponse)
async def quote_fetch(request: Request):
    try:
        quotes , status = read_all_quotes()

        if not status:
            return JSONResponse(content={"quote_available": False },status_code=200)
        
        if isinstance(quotes, list) and len(quotes) > 0:
            return JSONResponse(content={"quote_available": True,"quotes":quotes},status_code=200)

        return JSONResponse(content={"quote_available": False},status_code=200)

    except Exception as e:
        exception_traceback = traceback.format_exc()
        error_name = type(e).__name__
        log_error(message=f"Location : quote_fetch - file : server.py : Error Name - {error_name}",exc_info=exception_traceback)
        return JSONResponse(content={"quote_available": False },status_code=200)
    
@app.post('/quote_post',response_class=JSONResponse)
async def quote_post(request: Request):
    try:
        access_token = request.cookies.get("access_token")
        if not access_token:
            return JSONResponse(content={"status": False, "reason": "Unauthorized"}, status_code=200)
        
        token_data = read_access_token(access_token)

        if not token_data.get("success"):
            response = JSONResponse(content={"status": False, "reason": "Unauthorized"}, status_code=200)
            response.delete_cookie("access_token")
            return response
        
        request_data = await request.json()
        quote = Quote(**request_data)
        quote , status = write_quote(quote)

        if not status:
            return JSONResponse(content={"status": False, "reason": quote}, status_code=200)
        
        return JSONResponse(content={"status": True, "quote": quote.model_dump() if isinstance(quote, Quote) else None}, status_code=200)
        

    except Exception as e:
        exception_traceback = traceback.format_exc()
        error_name = type(e).__name__
        log_error(message=f"Location : quote_post - file : server.py : Error Name - {error_name}",exc_info=exception_traceback)
        return JSONResponse(content={ "status": False, "reason": "Internal server error"}, status_code=200)
    
@app.post('/logout',response_class=JSONResponse)
async def logout(request: Request):
    response = JSONResponse(content={"status": True, "message": "Logged out successfully"}, status_code=200)
    response.delete_cookie("access_token")
    return response

@app.post('/admin_quote_get',response_class=JSONResponse)
async def admin_quote_get(request: Request):
    try:
        access_token = request.cookies.get("access_token")
        if not access_token:
            return JSONResponse(content={"status": False, "reason": "Unauthorized"}, status_code=200)
        
        token_data = read_access_token(access_token)

        if not token_data.get("success"):
            response = JSONResponse(content={"status": False, "reason": "Unauthorized"}, status_code=200)
            response.delete_cookie("access_token")
            return response
        
        author = token_data["username"]
        quotes , status = read_quotes_by_author(author)
        if not status:
            return JSONResponse(content={"quote_available": False , "reason": "Failed to fetch quotes"},status_code=200)
        
        if isinstance(quotes, list) and len(quotes) > 0:
            return JSONResponse(content={"quote_available": True,"quotes":quotes},status_code=200)

        return JSONResponse(content={"quote_available": False , "reason": "No quotes found for the specified author"},status_code=200)

    except Exception as e:
        exception_traceback = traceback.format_exc()
        error_name = type(e).__name__
        log_error(message=f"Location : admin_quote_get - file : server.py : Error Name - {error_name}",exc_info=exception_traceback)
        return JSONResponse(content={"quote_available": False , "reason": "An error occurred while fetching quotes"},status_code=200)
    
@app.post('/quote_delete',response_class=JSONResponse)
async def quote_delete(request: Request):
    try:
        access_token = request.cookies.get("access_token")
        if not access_token:
            return JSONResponse(content={"status": False, "reason": "Unauthorized"}, status_code=200)
        
        token_data = read_access_token(access_token)

        if not token_data.get("success"):
            response = JSONResponse(content={"status": False, "reason": "Unauthorized"}, status_code=200)
            response.delete_cookie("access_token")
            return response
        
        request_data = await request.json()
        quote_id = request_data.get("quote_id")

        if not quote_id:
            return JSONResponse(content={"deleted": False, "reason": "Quote ID is required"}, status_code=200)
        
        message , status = delete_quote_by_id(quote_id)

        if not status:
            return JSONResponse(content={"deleted": False, "reason": message}, status_code=200)
        
        if message is None:
            return JSONResponse(content={"deleted": True, "message": "Quote deleted successfully"}, status_code=200)
        else:
            return JSONResponse(content={"deleted": False, "reason": message}, status_code=200)

    except Exception as e:
        exception_traceback = traceback.format_exc()
        error_name = type(e).__name__
        log_error(message=f"Location : quote_delete - file : server.py : Error Name - {error_name}",exc_info=exception_traceback)
        return JSONResponse(content={"deleted": False, "reason": "An error occurred while deleting the quote"}, status_code=200)