| | from enum import Enum |
| | import time |
| | import traceback |
| | import uuid |
| | from werkzeug.exceptions import InternalServerError |
| | from config.logger import CustomLogger,request_id_var |
| | import torch |
| | from transformers import pipeline, AutoModelForSequenceClassification, AutoTokenizer |
| | import os |
| | import sys |
| | import nltk |
| | from nltk.tokenize.punkt import PunktSentenceTokenizer |
| |
|
| | try: |
| | if getattr(sys, 'frozen', False): |
| | application_path = sys._MEIPASS |
| | else: |
| | application_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..') |
| | log=CustomLogger() |
| | log.info(f"application_path : {application_path}") |
| | log.info("before loading model") |
| | |
| | device = "cuda" |
| | device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu") |
| | log.info(f"device : {device}") |
| | gpu=0 if torch.cuda.is_available() else -1 |
| |
|
| | |
| | pipeline_kwargs={"return_token_type_ids": False,"max_length": 512,"truncation": True,"batch_size": 1} |
| | gibberishModel = AutoModelForSequenceClassification.from_pretrained(os.path.join(application_path, "models/gibberish")).to(device) |
| | gibberishTokenizer = AutoTokenizer.from_pretrained(os.path.join(application_path, "models/gibberish")) |
| | |
| | request_id_var.set("Startup") |
| | log_dict={} |
| | log.info("model loaded") |
| |
|
| | except Exception as e: |
| | log.error(f"Exception: {e}") |
| | log.error(f"Exception: {str(traceback.extract_tb(e.__traceback__)[0].lineno),e}") |
| |
|
| |
|
| |
|
| | class MatchType(Enum): |
| | SENTENCE = "sentence" |
| | FULL = "full" |
| |
|
| | def get_inputs(self, prompt: str) -> list[str]: |
| | if self == MatchType.SENTENCE: |
| | pk = PunktSentenceTokenizer() |
| | return pk.sentences_from_text(text=prompt) |
| | return [prompt] |
| | |
| |
|
| | class Gibberish: |
| |
|
| | def scan(self,payload): |
| | log.info("inside gibberish_check") |
| | id=uuid.uuid4().hex |
| | request_id_var.set(id) |
| | log_dict[request_id_var.get()]=[] |
| | try: |
| | st = time.time() |
| | text=payload['text'] |
| | gibberish_labels = payload['labels'] |
| | nlp = pipeline(task="text-classification", model=gibberishModel, tokenizer=gibberishTokenizer, device=device,model_kwargs=pipeline_kwargs) |
| | match_type = MatchType(MatchType.FULL) |
| | results_all = nlp(match_type.get_inputs(text)) |
| | log.debug(f"Gibberish detection finished :{results_all}") |
| | output={} |
| | res=[] |
| | for result in results_all: |
| | score = round( |
| | result["score"] if result["label"] in gibberish_labels else 1 - result["score"], |
| | 2, |
| | ) |
| | output['gibberish_label'] = result["label"] |
| | output['gibberish_score'] = score |
| |
|
| | res.append(output) |
| |
|
| | del nlp |
| | er=log_dict[request_id_var.get()] |
| | logobj = {"_id":id,"error":er} |
| | if len(er)!=0: |
| | log.debug(str(logobj)) |
| | del log_dict[id] |
| | return {"result":res,"time_taken":str(round(time.time()-st,3))+"s"} |
| | |
| | except Exception as e: |
| | log.error("Error occured in gibberish_check") |
| | log.error(f"Exception: {str(traceback.extract_tb(e.__traceback__)[0].lineno),e}") |
| | log_dict[request_id_var.get()].append({"Line number":str(traceback.extract_tb(e.__traceback__)[0].lineno),"Error":str(e), |
| | "Error Module":"Failed at gibberish_check call"}) |
| | raise InternalServerError() |
