Noveramaaz/Text_classification_API_Docker

1from contextlib import asynccontextmanager2from fastapi import FastAPI, HTTPException3from pydantic import BaseModel, ValidationError4from fastapi.encoders import jsonable_encoder5 6# TEXT PREPROCESSING7# --------------------------------------------------------------------8import re9import string10import nltk11nltk.download('punkt')12nltk.download('wordnet')13nltk.download('omw-1.4')14from nltk.stem import WordNetLemmatizer15 16# Function to remove URLs from text17def remove_urls(text):18    return re.sub(r'http[s]?://\S+', '', text)19 20# Function to remove punctuations from text21def remove_punctuation(text):22    regular_punct = string.punctuation23    return str(re.sub(r'['+regular_punct+']', '', str(text)))24 25# Function to convert the text into lower case26def lower_case(text):27    return text.lower()28 29# Function to lemmatize text30def lemmatize(text):31    wordnet_lemmatizer = WordNetLemmatizer()32 33    tokens = nltk.word_tokenize(text)34    lemma_txt = ''35    for w in tokens:36        lemma_txt = lemma_txt + wordnet_lemmatizer.lemmatize(w) + ' '37 38    return lemma_txt39 40def preprocess_text(text):41    # Preprocess the input text42    text = remove_urls(text)43    text = remove_punctuation(text)44    text = lower_case(text)45    text = lemmatize(text)46    return text47 48# Load the model using FastAPI lifespan event so that the model is loaded at the beginning for efficiency49@asynccontextmanager50async def lifespan(app: FastAPI):51    # Load the model from HuggingFace transformers library52    from transformers import pipeline53    global sentiment_task54    sentiment_task = pipeline("sentiment-analysis", model="cardiffnlp/twitter-roberta-base-sentiment-latest", tokenizer="cardiffnlp/twitter-roberta-base-sentiment-latest")55    yield56    # Clean up the model and release the resources57    del sentiment_task58 59# Initialize the FastAPI app60app = FastAPI(lifespan=lifespan)61 62# Define the input data model63class TextInput(BaseModel):64    text: str65 66# Define the welcome endpoint67@app.get('/')68async def welcome():69    return "Welcome to our Text Classification API"70 71# Validate input text length72MAX_TEXT_LENGTH = 100073 74# Define the sentiment analysis endpoint 75@app.post('/analyze/{text}')76async def classify_text(text_input:TextInput):    77    try:78        # Convert input data to JSON serializable dictionary79        text_input_dict = jsonable_encoder(text_input)80        # Validate input data using Pydantic model81        text_data = TextInput(**text_input_dict)  # Convert to Pydantic model82 83        # Validate input text length84        if len(text_input.text) > MAX_TEXT_LENGTH:85            raise HTTPException(status_code=400, detail="Text length exceeds maximum allowed length")86        elif len(text_input.text) == 0:87            raise HTTPException(status_code=400, detail="Text cannot be empty")88    except ValidationError as e:89        # Handle validation error90        raise HTTPException(status_code=422, detail=str(e))91 92    try:93        # Perform text classification94        return sentiment_task(preprocess_text(text_input.text))95    except ValueError as ve:96        # Handle value error97        raise HTTPException(status_code=400, detail=str(ve))98    except Exception as e:99        # Handle other server errors100        raise HTTPException(status_code=500, detail=str(e))101 102# Load the model using FastAPI lifespan event so that the model is loaded at the beginning for efficiency103@asynccontextmanager104async def lifespan(app: FastAPI):105    # Load the model from HuggingFace transformers library106    from transformers import pipeline107    global sentiment_task108    sentiment_task = pipeline("sentiment-analysis", model="cardiffnlp/twitter-roberta-base-sentiment-latest", tokenizer="cardiffnlp/twitter-roberta-base-sentiment-latest")109    yield110    # Clean up the model and release the resources111    del sentiment_task112 113# Initialize the FastAPI app114app = FastAPI(lifespan=lifespan)115 116# Define the input data model117class TextInput(BaseModel):118    text: str119 120# Define the welcome endpoint121@app.get('/')122async def welcome():123    return "Welcome to our Text Classification API"124 125# Validate input text length126MAX_TEXT_LENGTH = 1000127 128# Define the sentiment analysis endpoint 129@app.post('/analyze/{text}')130async def classify_text(text_input:TextInput):    131    try:132        # Convert input data to JSON serializable dictionary133        text_input_dict = jsonable_encoder(text_input)134        # Validate input data using Pydantic model135        text_data = TextInput(**text_input_dict)  # Convert to Pydantic model136 137        # Validate input text length138        if len(text_input.text) > MAX_TEXT_LENGTH:139            raise HTTPException(status_code=400, detail="Text length exceeds maximum allowed length")140        elif len(text_input.text) == 0:141            raise HTTPException(status_code=400, detail="Text cannot be empty")142    except ValidationError as e:143        # Handle validation error144        raise HTTPException(status_code=422, detail=str(e))145 146    try:147        # Perform text classification148        return sentiment_task(preprocess_text(text_input.text))149    except ValueError as ve:150        # Handle value error151        raise HTTPException(status_code=400, detail=str(ve))152    except Exception as e:153        # Handle other server errors154        raise HTTPException(status_code=500, detail=str(e))