File size: 11,344 Bytes

347eef9

# 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)}")