Update app.py

app.py

@@ -1,549 +1,16 @@
-# from fastapi import FastAPI, HTTPException, Depends, status
-# from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
-# from pydantic import BaseModel
-# from typing import List, Optional
-# from datetime import datetime, timedelta
-# import jwt
-# from passlib.context import CryptContext
-
-# app = FastAPI()
-
-# # JWT Configuration
-# SECRET_KEY = "123456"  # Change this to a secure random key
-# ALGORITHM = "HS256"
-# ACCESS_TOKEN_EXPIRE_MINUTES = 30
-
-# # Password hashing
-# pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-
-# # Security scheme
-# security = HTTPBearer()
-
-# # Models
-# class Student(BaseModel):
-#     rollno: int
-#     name: str
-#     age: int
-#     phone: str
-
-# class User(BaseModel):
-#     username: str
-#     email: str
-
-# class UserCreate(BaseModel):
-#     username: str
-#     email: str
-#     password: str
-
-# class UserLogin(BaseModel):
-#     username: str
-#     password: str
-
-# class Token(BaseModel):
-#     access_token: str
-#     token_type: str
-
-# # In-memory databases
-# Student_db: List[Student] = []
-# users_db: List[dict] = []  # Store users with hashed passwords
-
-# # Utility functions
-# def hash_password(password: str) -> str:
-#     return pwd_context.hash(password)
-
-# def verify_password(plain_password: str, hashed_password: str) -> bool:
-#     return pwd_context.verify(plain_password, hashed_password)
-
-# def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
-#     to_encode = data.copy()
-#     if expires_delta:
-#         expire = datetime.utcnow() + expires_delta
-#     else:
-#         expire = datetime.utcnow() + timedelta(minutes=15)
-#     to_encode.update({"exp": expire})
-#     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-#     return encoded_jwt
-
-# def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
-#     token = credentials.credentials
-#     try:
-#         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
-#         username: str = payload.get("sub")
-#         if username is None:
-#             raise HTTPException(
-#                 status_code=status.HTTP_401_UNAUTHORIZED,
-#                 detail="Invalid authentication credentials",
-#                 headers={"WWW-Authenticate": "Bearer"},
-#             )
-#         return username
-#     except jwt.PyJWTError:
-#         raise HTTPException(
-#             status_code=status.HTTP_401_UNAUTHORIZED,
-#             detail="Invalid authentication credentials",
-#             headers={"WWW-Authenticate": "Bearer"},
-#         )
-
-# def get_user_by_username(username: str):
-#     for user in users_db:
-#         if user["username"] == username:
-#             return user
-#     return None
-
-# # Authentication endpoints
-# @app.post("/register", response_model=dict)
-# def register_user(user: UserCreate):
-#     # Check if user already exists
-#     if get_user_by_username(user.username):
-#         raise HTTPException(
-#             status_code=400,
-#             detail="Username already registered"
-#         )
-    
-#     # Hash password and store user
-#     hashed_password = hash_password(user.password)
-#     user_dict = {
-#         "username": user.username,
-#         "email": user.email,
-#         "hashed_password": hashed_password
-#     }
-#     users_db.append(user_dict)
-    
-#     return {"message": "User registered successfully"}
-
-# @app.post("/login", response_model=Token)
-# def login_user(user_credentials: UserLogin):
-#     user = get_user_by_username(user_credentials.username)
-    
-#     if not user or not verify_password(user_credentials.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=ACCESS_TOKEN_EXPIRE_MINUTES)
-#     access_token = create_access_token(
-#         data={"sub": user["username"]}, expires_delta=access_token_expires
-#     )
-    
-#     return {"access_token": access_token, "token_type": "bearer"}
-
-# # Protected student endpoints
-# @app.post("/Student/", response_model=Student)
-# def create_student(student: Student, current_user: str = Depends(verify_token)):
-#     for existing in Student_db:
-#         if existing.rollno == student.rollno:
-#             raise HTTPException(status_code=400, detail="Student roll number already exists")
-#     Student_db.append(student)
-#     return student
-
-# @app.get("/Student/", response_model=List[Student])
-# def get_all_student_details(current_user: str = Depends(verify_token)):
-#     return Student_db
-
-# @app.get("/Student/{rollno}", response_model=Student)
-# def get_student_details(rollno: int, current_user: str = Depends(verify_token)):
-#     for student in Student_db:
-#         if student.rollno == rollno:
-#             return student
-#     raise HTTPException(status_code=404, detail="Student not found")
-
-# @app.put("/Student/{rollno}", response_model=Student)
-# def update_student(rollno: int, updated_student: Student, current_user: str = Depends(verify_token)):
-#     for index, student in enumerate(Student_db):
-#         if student.rollno == rollno:
-#             Student_db[index] = updated_student
-#             return updated_student
-#     raise HTTPException(status_code=404, detail="Student not found")
-
-# @app.delete("/Student/{rollno}")
-# def delete_student(rollno: int, current_user: str = Depends(verify_token)):
-#     for index, student in enumerate(Student_db):
-#         if student.rollno == rollno:
-#             del Student_db[index]
-#             return {"message": "Student deleted successfully"}
-#     raise HTTPException(status_code=404, detail="Student not found")
-
-# # Protected user info endpoint
-# @app.get("/me", response_model=User)
-# def get_current_user_info(current_user: str = Depends(verify_token)):
-#     user = get_user_by_username(current_user)
-#     if not user:
-#         raise HTTPException(status_code=404, detail="User not found")
-#     return User(username=user["username"], email=user["email"])
-
-
-
-# from fastapi import FastAPI, HTTPException, Depends, status, Request
-# from pydantic import BaseModel
-# from typing import List
-# import json
-# import os
-# import random
-# import string
-
-# app = FastAPI()
-
-# VALID_USERNAME = "Jignesh"
-# VALID_PASSWORD = "123456"
-
-# active_tokens = set()
-
-# class Login(BaseModel):
-#     username: str
-#     password: str
-
-# class Student(BaseModel):
-#     rollno: int
-#     name: str
-#     age: int
-#     phone: str
-
-# def create_random_token() -> str:
-#     length = random.randint(32, 64)
-#     characters = string.ascii_letters + string.digits + '-_'
-#     return ''.join(random.choice(characters) for _ in range(length))
-
-# def validate_token(token: str) -> bool:
-#     return token in active_tokens
-
-# def load_data():
-#     if os.path.exists("students.json"):
-#         with open("students.json", "r") as f:
-#             return json.load(f)
-#     return []
-
-# def save_data():
-#     with open("students.json", "w") as f:
-#         json.dump(students, f, indent=2)
-
-# students = load_data()
-
-# @app.post("/login/")
-# def login(user: Login):
-#     if user.username != VALID_USERNAME or user.password != VALID_PASSWORD:
-#         raise HTTPException(status_code=400, detail="Invalid credentials")
-    
-#     access_token = create_random_token()
-#     active_tokens.add(access_token)
-#     return {"access_token": access_token, "token_type": "bearer"}
-
-# def get_current_user(request: Request):
-#     token = request.headers.get("authorization")
-#     if not token:
-#         raise HTTPException(
-#             status_code=status.HTTP_401_UNAUTHORIZED,
-#             detail="Not authenticated",
-#             headers={"WWW-Authenticate": "Bearer"},
-#         )
-    
-#     if not validate_token(token):
-#         raise HTTPException(
-#             status_code=status.HTTP_401_UNAUTHORIZED,
-#             detail="Invalid or expired token",
-#             headers={"WWW-Authenticate": "Bearer"},
-#         )
-#     return True
-
-# @app.post("/students/")
-# def create_student(student: Student, current_user: bool = Depends(get_current_user)):
-#     if any(s["rollno"] == student.rollno for s in students):
-#         raise HTTPException(status_code=400, detail="Roll number already exists")
-    
-#     students.append(student.dict())
-#     save_data()
-#     return {"message": "Student added successfully", "student": student}
-
-# @app.get("/students/")
-# def get_all_students(current_user: bool = Depends(get_current_user)):
-#     return students
-
-# @app.get("/students/{rollno}")
-# def get_student(rollno: int, current_user: bool = Depends(get_current_user)):
-#     student = next((s for s in students if s["rollno"] == rollno), None)
-#     if not student:
-#         raise HTTPException(status_code=404, detail="Student not found")
-#     return student
-
-# @app.put("/students/{rollno}")
-# def update_student(rollno: int, student: Student, current_user: bool = Depends(get_current_user)):
-#     for i, s in enumerate(students):
-#         if s["rollno"] == rollno:
-#             students[i] = student.dict()
-#             save_data()
-#             return {"message": "Student updated", "student": student}
-#     raise HTTPException(status_code=404, detail="Student not found")
-
-# @app.delete("/students/{rollno}")
-# def delete_student(rollno: int, current_user: bool = Depends(get_current_user)):
-#     for i, s in enumerate(students):
-#         if s["rollno"] == rollno:
-#             students.pop(i)
-#             save_data()
-#             return {"message": "Student deleted"}
-#     raise HTTPException(status_code=404, detail="Student not found")
-
-# import time
-# from fastapi import FastAPI, HTTPException, Request, status, Depends
-# from fastapi.security.api_key import APIKeyHeader
-# from pydantic import BaseModel
-# import json
-# import os
-
-# app = FastAPI()
-
-# # Authentication configuration
-# DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0"
-# api_key_header = APIKeyHeader(name="Token", auto_error=False)
-
-# def verify_token(token: str = Depends(api_key_header)):
-#     if token != DEFAULT_TOKEN:
-#         raise HTTPException(
-#             status_code=status.HTTP_401_UNAUTHORIZED,
-#             detail="Invalid or missing token. Use correct Token"
-#         )
-
-# # Middleware to add process time header
-# @app.middleware("http")
-# async def add_process_time_header(request: Request, call_next):
-#     start_time = time.perf_counter()
-#     response = await call_next(request)
-#     process_time = time.perf_counter() - start_time
-#     response.headers["Time of pro"] = str(process_time)
-#     return response
-
-# class Student(BaseModel):
-#     rollno: int
-#     name: str
-#     age: int
-#     phone: str
-
-# def load_data():
-#     if os.path.exists("students.json"):
-#         with open("students.json", "r") as f:
-#             return json.load(f)
-#     return []
-
-# def save_data():
-#     with open("students.json", "w") as f:
-#         json.dump(students, f, indent=2)
-
-# students = load_data()
-
-# @app.post("/Student/", dependencies=[Depends(verify_token)])
-# def create_student(student: Student):
-#     if any(s["rollno"] == student.rollno for s in students):
-#         raise HTTPException(status_code=400, detail="Roll number already exists")
-    
-#     students.append(student.dict())
-#     save_data()
-#     return student
-
-# @app.get("/Student/", dependencies=[Depends(verify_token)])
-# def get_all_students():
-#     return students
-
-# @app.get("/Student/{rollno}", dependencies=[Depends(verify_token)])
-# def get_student(rollno: int):
-#     student = next((s for s in students if s["rollno"] == rollno), None)
-#     if not student:
-#         raise HTTPException(status_code=404, detail="Student not found")
-#     return student
-
-# @app.put("/Student/{rollno}", dependencies=[Depends(verify_token)])
-# def update_student(rollno: int, student: Student):
-#     for i, s in enumerate(students):
-#         if s["rollno"] == rollno:
-#             students[i] = student.dict()
-#             save_data()
-#             return student
-#     raise HTTPException(status_code=404, detail="Student not found")
-
-# @app.delete("/Student/{rollno}", dependencies=[Depends(verify_token)])
-# def delete_student(rollno: int):
-#     for i, s in enumerate(students):
-#         if s["rollno"] == rollno:
-#             students.pop(i)
-#             save_data()
-#             return {"message": "Student deleted"}
-#     raise HTTPException(status_code=404, detail="Student not found")
-
-
-
-# import time
-# import joblib
-# import numpy as np
-# from fastapi import FastAPI, HTTPException, Request, status, Depends
-# from fastapi.security.api_key import APIKeyHeader
-# from pydantic import BaseModel
-
-# app = FastAPI()
-# DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0"
-# api_key_header = APIKeyHeader(name="Token", auto_error=False)
-
-# def verify_token(token: str = Depends(api_key_header)):
-#     if token != DEFAULT_TOKEN:
-#         raise HTTPException(
-#             status_code=status.HTTP_401_UNAUTHORIZED,
-#             detail="Invalid or missing token. Use correct Token"
-#         )
-
-# @app.middleware("http")
-# async def add_process_time_header(request: Request, call_next):
-#     start_time = time.perf_counter()
-#     response = await call_next(request)
-#     process_time = time.perf_counter() - start_time
-#     response.headers["Time of pro"] = str(process_time)
-#     return response
-
-# model = [joblib.load(r"D:\Igenerate_code\Model\logistic_regression.pkl"),
-#             joblib.load(r"D:\Igenerate_code\Model\random_forest_model.pkl"),
-#             joblib.load(r"D:\Igenerate_code\Model\decision_tree_model.pkl"),
-#             joblib.load(r"D:\Igenerate_code\Model\SVM_model.pkl"),
-#             joblib.load(r"D:\Igenerate_code\Model\KNeighborsClassifier_model.pkl"),
-#             joblib.load(r"D:\Igenerate_code\Model\Naive_Bayes_model.pkl"),
-#             joblib.load(r"D:\Igenerate_code\Model\ANN_model.pkl")]
-
-# class MusicFeatures(BaseModel):
-#     danceability: float
-#     energy: float
-#     key: int
-#     loudness: float
-#     mode: int
-#     speechiness: float
-#     acousticness: float
-#     instrumentalness: float
-#     liveness: float
-#     valence: float
-#     tempo: float
-#     duration_ms: int
-#     time_signature: int
-
-# @app.post("/predict", dependencies=[Depends(verify_token)])
-# def predict_music(features: MusicFeatures):
-#     try:
-#         input_data = np.array([[  # Format input as 2D array
-#             features.danceability, features.energy, features.key,
-#             features.loudness, features.mode, features.speechiness,
-#             features.acousticness, features.instrumentalness, features.liveness,
-#             features.valence, features.tempo, features.duration_ms, features.time_signature
-
-#         ]])
-#         prediction = model.predict(input_data)[0]
-#         return {"prediction": int(prediction)}
-#     except Exception as e:
-#         raise HTTPException(status_code=500, detail=str(e))
-
-
-
-# import time
-# import joblib
-# import numpy as np
-# from fastapi import FastAPI, HTTPException, Request, status, Depends
-# from fastapi.security.api_key import APIKeyHeader
-# from pydantic import BaseModel
-
-# app = FastAPI()
-
-# DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0"
-# api_key_header = APIKeyHeader(name="Token", auto_error=False)
-
-# def verify_token(token: str = Depends(api_key_header)):
-#     if token != DEFAULT_TOKEN:
-#         raise HTTPException(
-#             status_code=status.HTTP_401_UNAUTHORIZED,
-#             detail="Invalid or missing token. Use correct Token"
-#         )
-
-# @app.middleware("http")
-# async def add_process_time_header(request: Request, call_next):
-#     start_time = time.perf_counter()
-#     response = await call_next(request)
-#     process_time = time.perf_counter() - start_time
-#     response.headers["Time of pro"] = str(process_time)
-#     return response
-
-# model_paths = {
-#     "logistic_regression": r"D:\Igenerate_code\Model\logistic_regression.pkl",
-#     "random_forest_model": r"D:\Igenerate_code\Model\random_forest_model.pkl",
-#     "DecisionTreeClassifier_model": r"D:\Igenerate_code\Model\DecisionTreeClassifier.pkl",
-#     "SVM_model": r"D:\Igenerate_code\Model\SVM_model.pkl",
-#     "KNeighborsClassifier_model": r"D:\Igenerate_code\Model\KNeighborsClassifier_model.pkl",
-#     "Naive_Bayes_model": r"D:\Igenerate_code\Model\Naive_Bayes_model.pkl",
-#     "ANN_model": r"D:\Igenerate_code\Model\ANN_model.pkl"
-# }
-
-# models = {name: joblib.load(path) for name, path in model_paths.items()}
-
-# class MusicFeatures(BaseModel):
-#     model_name: "random_forest_model" 
-#     danceability: float
-#     energy: float
-#     key: int
-#     loudness: float
-#     mode: int
-#     speechiness: float
-#     acousticness: float
-#     instrumentalness: float
-#     liveness: float
-#     valence: float
-#     tempo: float
-#     duration_ms: int
-#     time_signature: int
-
-
-# @app.post("/predict", dependencies=[Depends(verify_token)])
-# def predict_music(features: MusicFeatures):
-#     try:
-#         if features.model_name not in models:
-#             raise HTTPException(status_code=400, detail="Invalid model name")
-
-#         model = models[features.model_name]
-
-#         input_data = np.array([[  
-#             features.danceability, features.energy, features.key,
-#             features.loudness, features.mode, features.speechiness,
-#             features.acousticness, features.instrumentalness, features.liveness,
-#             features.valence, features.tempo, features.duration_ms, features.time_signature
-#         ]])
-
-#         prediction = model.predict(input_data)[0]
-#         return {
-#             "model_used": features.model_name,
-#             "prediction": int(prediction)
-#         }
-
-#     except Exception as e:
-#         raise HTTPException(status_code=500, detail=str(e))
-
 import time
 import joblib
 import numpy as np
-from fastapi import FastAPI, HTTPException, Request, status, Depends
-from fastapi.security.api_key import APIKeyHeader
+from fastapi import FastAPI, HTTPException, Request
 from pydantic import BaseModel
 from fastapi.templating import Jinja2Templates
 from fastapi.responses import HTMLResponse
-
-
 from fastapi.staticfiles import StaticFiles
 
-
 app = FastAPI()
 templates = Jinja2Templates(directory="templates")
 app.mount("/static", StaticFiles(directory="static"), name="static")
 
-
-DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0"
-api_key_header = APIKeyHeader(name="Token", auto_error=False)
-
-def verify_token(token: str = Depends(api_key_header)):
-    if token != DEFAULT_TOKEN:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid or missing token. Use correct Token"
-        )
-
 @app.middleware("http")
 async def add_process_time_header(request: Request, call_next):
     start_time = time.perf_counter()
@@ -588,35 +55,34 @@ def make_prediction(model, features: MusicFeatures):
     ]])
     return int(model.predict(input_data)[0])
 
-
 @app.get("/", response_class=HTMLResponse)
 async def read_home(request: Request):
     return templates.TemplateResponse("index.html", {"request": request})
 
-@app.post("/predict/logistic_regression", dependencies=[Depends(verify_token)])
+@app.post("/predict/logistic_regression")
 def predict_logistic(features: MusicFeatures):
     return {"model": "Logistic Regression", "prediction": make_prediction(models["logistic_regression"], features)}
 
-@app.post("/predict/random_forest", dependencies=[Depends(verify_token)])
+@app.post("/predict/random_forest")
 def predict_rf(features: MusicFeatures):
     return {"model": "Random Forest", "prediction": make_prediction(models["random_forest_model"], features)}
 
-@app.post("/predict/decision_tree", dependencies=[Depends(verify_token)])
+@app.post("/predict/decision_tree")
 def predict_dt(features: MusicFeatures):
     return {"model": "Decision Tree", "prediction": make_prediction(models["decision_tree"], features)}
 
-@app.post("/predict/svm", dependencies=[Depends(verify_token)])
+@app.post("/predict/svm")
 def predict_svm(features: MusicFeatures):
     return {"model": "SVM", "prediction": make_prediction(models["svm"], features)}
 
-@app.post("/predict/knn", dependencies=[Depends(verify_token)])
+@app.post("/predict/knn")
 def predict_knn(features: MusicFeatures):
     return {"model": "K-Nearest Neighbors", "prediction": make_prediction(models["knn"], features)}
 
-@app.post("/predict/naive_bayes", dependencies=[Depends(verify_token)])
+@app.post("/predict/naive_bayes")
 def predict_nb(features: MusicFeatures):
     return {"model": "Naive Bayes", "prediction": make_prediction(models["naive_bayes"], features)}
 
-@app.post("/predict/ann", dependencies=[Depends(verify_token)])
+@app.post("/predict/ann")
 def predict_ann(features: MusicFeatures):
     return {"model": "Artificial Neural Network", "prediction": make_prediction(models["ann"], features)}