# import os # import re # from aiohttp import ClientSession # from google.cloud import vision # from io import BytesIO # import google.generativeai as genai # import traceback # import nltk # from nltk.corpus import stopwords # from nltk.tokenize import word_tokenize # from nltk.stem import WordNetLemmatizer # from string import punctuation # from PIL import Image # import io # from PIL import Image, ImageEnhance, ImageFilter # # Download necessary NLTK resources # nltk.download('punkt') # nltk.download('stopwords') # nltk.download('wordnet') # os.environ['GOOGLE_API_KEY'] = "AIzaSyA9sqz4YKQHKXR9TU1imw0DPOghzHOMiBo" # # genai.configure(api_key = os.environ['GOOGLE_API_KEY']) # os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = 'cloudVisionAPI.json' # # model = genai.GenerativeModel('gemini-pro') # # model = genai.GenerativeModel('gemini-1.5-flash') # async def remove_text_from_field(texts_to_remove, text_field): # for pattern in texts_to_remove: # text_field = re.sub(pattern, "", text_field) # return text_field # async def getImage(img_url): # try: # async with ClientSession() as session: # async with session.get(img_url) as response: # img_data = await response.read() # return BytesIO(img_data) # except Exception as e: # raise ValueError(f"Error in getImage: {str(e)}") # # async def detectText(url): # # try: # # client = vision.ImageAnnotatorClient() # # image_bytes = await getImage(url) # # image = vision.Image(content=image_bytes.getvalue()) # # image_response = client.document_text_detection(image=image) # # if image_response.error.message: # # raise Exception("{}\nFor more info on error messages, check: ""https://cloud.google.com/apis/design/errors".format(image_response.error.message)) # # image_texts = image_response.text_annotations # # imageData = image_texts[0].description # # return imageData # # except Exception as e: # # traceback.print_exc() # # raise ValueError(f"Error in detectText: {str(e)}") # async def detectText(url, threshold=0.0): # try: # client = vision.ImageAnnotatorClient() # image_bytes = await getImage(url) # image = vision.Image(content=image_bytes.getvalue()) # image_response = client.document_text_detection(image=image) # if image_response.error.message: # raise Exception( # "{}\nFor more info on error messages, check: " # "https://cloud.google.com/apis/design/errors".format(image_response.error.message) # ) # image_texts = image_response.full_text_annotation # bangla_words = [] # for page in image_texts.pages: # for block in page.blocks: # for paragraph in block.paragraphs: # for word in paragraph.words: # word_text = ''.join([symbol.text for symbol in word.symbols]) # if word.confidence >= threshold: # # Check if the word contains any Bangla character # if re.search(r'[\u0980-\u09FF]', word_text): # bangla_words.append(word_text) # return ' '.join(bangla_words) # except Exception as e: # traceback.print_exc() # raise ValueError(f"Error in detectText: {str(e)}") # def clean_text(text): # tokens = word_tokenize(text) # tokens = [word for word in tokens if word not in punctuation] # lemmatizer = WordNetLemmatizer() # tokens = [lemmatizer.lemmatize(word) for word in tokens] # clean_tokens = [i for i in tokens if i != '·'] # print("Cleaned Tokens:", clean_tokens) # return ' '.join(clean_tokens) # async def main(url): # try: # data = await detectText(url) # # myQue = f"Extract only name from {data} also correct the name of Cigarettes and person name if the name is wrong. Dont give any other information except those name." # # response = model.generate_content(myQue) # # text = response.text # cleaned_text = clean_text(data) # return cleaned_text # except Exception as e: # traceback.print_exc() # raise ValueError(f"Error in main: {str(e)}") import os import re import io import traceback import nltk from aiohttp import ClientSession from google.cloud import vision from io import BytesIO from string import punctuation from PIL import Image, ImageEnhance, ImageFilter, ImageDraw, ImageFont from nltk.corpus import stopwords from nltk.tokenize import word_tokenize from nltk.stem import WordNetLemmatizer # # Download necessary NLTK resources # nltk.download('punkt') # nltk.download('stopwords') # nltk.download('wordnet') # Environment variables os.environ['GOOGLE_API_KEY'] = "AIzaSyA9sqz4YKQHKXR9TU1imw0DPOghzHOMiBo" os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = 'cloudVisionAPI.json' # ========== Utility Functions ========== async def getImage(img_url): try: async with ClientSession() as session: async with session.get(img_url) as response: img_data = await response.read() return BytesIO(img_data) except Exception as e: raise ValueError(f"Error in getImage: {str(e)}") # async def preprocess_image_for_ocr(image_bytes): # image = Image.open(io.BytesIO(image_bytes)).convert('L') # contrast = ImageEnhance.Contrast(image) # image = contrast.enhance(2.0) # image = image.point(lambda x: 0 if x < 140 else 255, '1') # image = image.filter(ImageFilter.SHARPEN) # image = image.filter(ImageFilter.MedianFilter()) # # Resize image (scale up) # new_size = (image.size[0] * 2, image.size[1] * 2) # image = image.resize(new_size) # # Generate thumbnail (optional) # thumb_size = (640, 640) # image.thumbnail(thumb_size) # # Convert PIL image to bytes for Google Vision # byte_arr = io.BytesIO() # image.save("preprocessed_output.png") # return byte_arr.getvalue() counter = 0 # async def preprocess_image_for_ocr(image_bytes): # image = Image.open(io.BytesIO(image_bytes)).convert('L') # Convert to grayscale # # Enhance contrast # contrast = ImageEnhance.Contrast(image) # image = contrast.enhance(2.0) # # Binarize image # image = image.point(lambda x: 0 if x < 140 else 255, '1') # # Apply filters # image = image.filter(ImageFilter.SHARPEN) # image = image.filter(ImageFilter.MedianFilter()) # # Resize image (scale up) # new_size = (image.size[0] * 2, image.size[1] * 2) # image = image.resize(new_size) # # # Generate thumbnail (optional) # thumb_size = (640, 640) # image.thumbnail(thumb_size) # # Convert PIL image to bytes for Google Vision # byte_arr = io.BytesIO(image.getvalue()) # # image.save(byte_arr, format='PNG') # Save to buffer instead of disk # return byte_arr.getvalue() async def preprocess_image_for_ocr(image_bytes): global counter # Since image_bytes is already a BytesIO object, we can directly use it image = Image.open(image_bytes).convert('L') # Convert to grayscale # Enhance contrast # contrast = ImageEnhance.Contrast(image) # image = contrast.enhance(2.0) # # Binarize image # image = image.point(lambda x: 0 if x < 140 else 255, '1') # # Apply filters # image = image.filter(ImageFilter.SHARPEN) # image = image.filter(ImageFilter.MedianFilter()) # # Resize image (scale up) # new_size = (image.size[0] * 2, image.size[1] * 2) # image = image.resize(new_size) # Save image to a bytes buffer byte_arr = io.BytesIO() image.save(byte_arr, format='PNG') counter +=1 image.save(f"images/{counter}_image.jpg", format='PNG') return byte_arr.getvalue() async def detectText(url, threshold=0.0): try: # Initialize the Vision client client = vision.ImageAnnotatorClient() # Fetch and preprocess image image_bytes = await getImage(url) processed_image_bytes = await preprocess_image_for_ocr(image_bytes) # Create Image object for Google Vision API image = vision.Image(content=processed_image_bytes) # Perform document text detection response = client.document_text_detection(image=image) # Check for errors in the API response if response.error.message: raise Exception(f"API Error: {response.error.message}\nCheck: https://cloud.google.com/apis/design/errors") # Extract text from the response image_texts = response.full_text_annotation bangla_words = [] # Loop through the detected text and filter Bangla words for page in image_texts.pages: for block in page.blocks: for paragraph in block.paragraphs: for word in paragraph.words: word_text = ''.join([symbol.text for symbol in word.symbols]) if word.confidence >= threshold and re.search(r'[\u0980-\u09FF]', word_text): bangla_words.append(word_text) # Return Bangla words as a space-separated string return ' '.join(bangla_words) except Exception as e: traceback.print_exc() raise ValueError(f"Error in detectText: {str(e)}") # async def detectText(url, threshold=0.0): # try: # client = vision.ImageAnnotatorClient() # image_bytes = await getImage(url) # processed_image_bytes = await preprocess_image_for_ocr(image_bytes.getvalue()) # image = vision.Image(content=processed_image_bytes) # response = client.document_text_detection(image=image) # if response.error.message: # raise Exception(f"{response.error.message}\nCheck: https://cloud.google.com/apis/design/errors") # image_texts = response.full_text_annotation # bangla_words = [] # for page in image_texts.pages: # for block in page.blocks: # for paragraph in block.paragraphs: # for word in paragraph.words: # word_text = ''.join([symbol.text for symbol in word.symbols]) # if word.confidence >= threshold and re.search(r'[\u0980-\u09FF]', word_text): # bangla_words.append(word_text) # return ' '.join(bangla_words) # except Exception as e: # traceback.print_exc() # raise ValueError(f"Error in detectText: {str(e)}") def clean_text(text): tokens = word_tokenize(text) tokens = [word for word in tokens if word not in punctuation] lemmatizer = WordNetLemmatizer() tokens = [lemmatizer.lemmatize(word) for word in tokens] clean_tokens = [i for i in tokens if i != '·'] print("Cleaned Tokens:", clean_tokens) return ' '.join(clean_tokens) async def main(url): try: data = await detectText(url) cleaned_text = clean_text(data) return cleaned_text except Exception as e: traceback.print_exc() raise ValueError(f"Error in main: {str(e)}")