Multimedika/Bot_Development

1import logging2from typing import Annotated, Optional3 4from api.function import data_ingestion5from api.router.user import user_dependency6from api.auth import check_user_authentication, check_admin_authentication7 8from fastapi import Form, APIRouter, File, UploadFile, Depends9from fastapi.responses import JSONResponse10 11from db.models import Metadata12from db.database import get_db13from db.query.query_book import BookQuery14from db.query.query_category import CategoryQuery15from db.fetching import DataFetching16 17from config import MYSQL_CONFIG18from utils.error_handlers import handle_exception19from script.vector_db import IndexManager20from service.dto import MetadataResponse21from service.aws_loader import Loader22from sqlalchemy.orm import Session23from sqlalchemy.future import select24 25 26router = APIRouter(tags=["Book"])27 28index_manager = IndexManager()29db_dependency = Annotated[Session, Depends(get_db)]30 31 32@router.get("/book")33async def get_metadata(user: user_dependency, db: db_dependency):34    auth_response = check_user_authentication(user)35    if auth_response:36        print(auth_response)37        return auth_response38    try:39        # Join Metadata with Category to get the category name40        fetching = DataFetching(user, db)41        metadata_fetching = fetching.metadata_fetching()42 43        # Transform results into MetadataResponse model with optional thumbnail handling44        return metadata_fetching45 46    except Exception as e:47        return handle_exception(e)48 49 50@router.post("/book")51async def upload_file(52    user: user_dependency,53    db: db_dependency,54    title: str = Form(...),55    author: str = Form(...),56    category_id: int = Form(...),57    year: int = Form(...),58    publisher: str = Form(...),59    file: UploadFile = File(...),60    lang: str = Form(None),61    thumbnail: Optional[UploadFile] = File(None),62):63    auth_response = check_admin_authentication(user)64    if auth_response:65        return auth_response66    67    # Restrict `lang` to only "id" or "en"68    lang = lang if lang in {"id", "en"} else "en"69 70    # Query the category based on category_id71    category_query = CategoryQuery(user)72    category = category_query.get_category(db, category_id)73 74    try:75        reference = {76            "title": title,77            "author": author,78            "category": category,79            "year": year,80            "publisher": publisher,81        }82 83        # Process the file and handle data ingestion84        response = await data_ingestion(reference, file, lang)85        86        if isinstance(response, JSONResponse):87            return response  # Return the error response directly88        89        if thumbnail:90            file_name = f"{reference['title']}"91            aws_loader = Loader()92            ekstensi_file = file.filename.split(".")[-1].lower()93            aws_loader.upload_image_to_s3(file=thumbnail, custom_name=f"{file_name}.{ekstensi_file}")94            95 96        # Create a new Metadata object97        book_query = BookQuery(user)98        book_query.add_book(db, title, author, category_id, year, publisher)99        logging.info("Database Inserted")100 101        return {102            "filename": file.filename,103            "response": response,104            "info": "upload file successfully",105        }106 107    except Exception as e:108        return handle_exception(e)109 110 111@router.put("/book/{metadata_id}")112async def update_metadata(113    user: user_dependency,114    db: db_dependency,115    metadata_id: int,116    title: str = Form(...),117    author: str = Form(...),118    category_id: int = Form(...),119    year: int = Form(...),120    publisher: str = Form(...),121    thumbnail: Optional[UploadFile] = File(None),122):123    auth_response = check_admin_authentication(user)124    if auth_response:125        return auth_response126 127    try:128        # fetch current metadata129        book_query = BookQuery(user)130        current_metadata = book_query.get_metadata_books(db, metadata_id)131 132        # Fetch current and new categories133        category_query = CategoryQuery(user)134        current_category = category_query.get_current_category(db, metadata_id)135        136        new_category = category_query.get_category(db, category_id)137 138 139        # Prepare the references140        current_reference = {141            "title": current_metadata.title,142            "author": current_metadata.author,143            "category": current_category,144            "year": current_metadata.year,145            "publisher": current_metadata.publisher,146        }147 148        new_reference = {149            "title": title,150            "author": author,151            "category": new_category,152            "year": year,153            "publisher": publisher,154        }155 156        # Update vector database157        index_manager.update_vector_database(current_reference, new_reference)158 159        # Update existing metadata entry160        metadata = db.query(Metadata).filter(Metadata.id == metadata_id).first()161        print(metadata)162 163        if not metadata:164            return JSONResponse(status_code=404, content="Metadata not found")165 166        updated_metadata = book_query.update_metadata_entry(167            db, metadata_id, title, author, category_id, year, publisher168        )169        print(updated_metadata)170        updated_category = category_query.get_category(db, updated_metadata.category_id)171        print(updated_category)172 173        return MetadataResponse(174            id=metadata_id,175            title=updated_metadata.title,176            author=updated_metadata.author,177            category=updated_category,  # Assuming category relationship is available178            category_id=updated_metadata.category_id,179            year=updated_metadata.year,180            publisher=updated_metadata.publisher,181            thumbnail=(182                updated_metadata.thumbnail183                if updated_metadata.thumbnail is not None184                else None185            ),186        )187 188    except Exception as e:189        return handle_exception(e)190 191@router.delete("/book/{metadata_id}")192async def delete_metadata(user: user_dependency, db: db_dependency, metadata_id: int):193    auth_response = check_admin_authentication(user)194    if auth_response:195        return auth_response196 197    try:198        # Check if metadata exists before deletion199        metadata = db.query(Metadata).filter(Metadata.id == metadata_id).first()200 201        if not metadata:202            return JSONResponse(status_code=404, content="Metadata not found")203 204        # Delete from the vector database and the database205        index_manager.delete_vector_database(metadata.title)206        207        db.delete(metadata)208        db.commit()209 210        return {"Status": "delete successfully"}211 212    except Exception as e:213        return handle_exception(e)214