tools/image_parser.py

import logging
import os
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
import easyocr

logger = logging.getLogger(__name__)

class ImageParserInput(BaseModel):
    task_id: str = Field(description="Task identifier")
    file_path: str = Field(description="Path to the image file")

async def image_parser_func(task_id: str, file_path: str) -> str:
    """
    Parse text from an image file using OCR.
    
    Args:
        task_id (str): Task identifier.
        file_path (str): Path to the image file.
    
    Returns:
        str: Extracted text or error message.
    """
    try:
        if not os.path.exists(file_path):
            logger.warning(f"Image file not found: {file_path}")
            return "Image file not found"
        
        logger.info(f"Parsing image: {file_path} for task {task_id}")
        reader = easyocr.Reader(['en'], model_storage_directory='./cache')
        result = reader.readtext(file_path, detail=0)
        text = " ".join(result).strip()
        return text if text else "No text extracted from image"
    except Exception as e:
        logger.error(f"Error parsing image for task {task_id}: {e}")
        return f"Error: {str(e)}"

image_parser_tool = StructuredTool.from_function(
    func=image_parser_func,
    name="image_parser_tool",
    args_schema=ImageParserInput,
    coroutine=image_parser_func
)