| # 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 | |
| import google.generativeai as genai | |
| # Environment variables | |
| os.environ['GOOGLE_API_KEY'] = "AIzaSyA9sqz4YKQHKXR9TU1imw0DPOghzHOMiBo" | |
| os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = 'cloudVisionAPI.json' | |
| genai.configure(api_key = os.environ['GOOGLE_API_KEY']) | |
| # # model = genai.GenerativeModel('gemini-pro') | |
| model = genai.GenerativeModel('gemini-1.5-flash') | |
| # ========== 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)}") | |
| counter = 0 | |
| async def preprocess_image_for_ocr(image_bytes): | |
| global counter | |
| image = Image.open(image_bytes).convert('L') # Convert to grayscale | |
| # 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)}") | |
| 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: | |
| dic = ['গোল্ডলিফ', 'লাকি স্ট্রাইক', 'বেনসন'] | |
| # Extract the data using detectText function | |
| data = await detectText(url) | |
| # Define the query (myQue) to pass to the model | |
| myQue = f""" | |
| Extract and correct the names of cigarette brands and Bangladeshi people's names from the following text. The text may contain spelling errors, grammatical issues, and improperly formatted names. | |
| Your task: | |
| - Extract and correct the names of cigarette brands and Bangladeshi people also bangla writings. | |
| - Return only the corrected data (in Bangla), with no additional information or explanations. | |
| Here is the provided text: | |
| {data} | |
| Only return the corrected names, no extra text. | |
| """ | |
| print("Query to Model:", myQue) # Optional: For debugging to see the query sent to the model | |
| # Generate the response from the model | |
| response = model.generate_content(myQue) | |
| text = response.text.strip() | |
| clean_txt = clean_text(text) | |
| print("Response Text:", clean_txt) | |
| # Return the cleaned response with only the extracted names | |
| return clean_txt | |
| except Exception as e: | |
| # Print and raise a detailed error message for debugging | |
| traceback.print_exc() | |
| raise ValueError(f"Error in main: {str(e)}") | |