python -m venv venv
source venv/bin/activate
pip install openai Pillow pytesseract httpx
export OPENAI_API_KEY="sk-your-key"

import base64
import json
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional

from openai import AsyncOpenAI

logger = logging.getLogger(__name__)


class GPT4VisionClient:
    """Production client for GPT-4 Vision API with async support."""

    def __init__(self, model: str = "gpt-4o"):
        self.client = AsyncOpenAI()
        self.model = model

    async def analyze_image(
        self,
        image_path: str,
        question: str = "Describe this image in detail.",
        detail: str = "high",
    ) -> str:
        base64_image = self._encode_image(image_path)
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": question},
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{base64_image}",
                                "detail": detail,
                            },
                        },
                    ],
                }
            ],
            max_tokens=1500,
        )
        return response.choices[0].message.content

    async def analyze_url(self, image_url: str, question: str = "Describe this image.") -> str:
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": question},
                        {"type": "image_url", "image_url": {"url": image_url, "detail": "high"}},
                    ],
                }
            ],
            max_tokens=1500,
        )
        return response.choices[0].message.content

    async def extract_structured_data(self, image_path: str, schema: str) -> Dict[str, Any]:
        base64_image = self._encode_image(image_path)
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": f"Extract data from this image into the following JSON schema:\n{schema}\n\nReturn ONLY valid JSON.",
                        },
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{base64_image}",
                                "detail": "high",
                            },
                        },
                    ],
                }
            ],
            max_tokens=2000,
        )
        try:
            return json.loads(response.choices[0].message.content)
        except json.JSONDecodeError:
            return {"raw": response.choices[0].message.content}

    def _encode_image(self, image_path: str) -> str:
        with open(image_path, "rb") as f:
            return base64.b64encode(f.read()).decode("utf-8")

    async def batch_analyze(self, image_paths: List[str], question: str) -> List[Dict[str, Any]]:
        import asyncio
        tasks = [self.analyze_image(path, question) for path in image_paths]
        answers = await asyncio.gather(*tasks)
        return [{"path": path, "answer": ans} for path, ans in zip(image_paths, answers)]

from typing import Any, Dict, List

import pytesseract
from PIL import Image


class OCRExtractor:
    """Extract text from images using Tesseract OCR with confidence scores."""

    def extract_text(self, image_path: str) -> str:
        image = Image.open(image_path)
        return pytesseract.image_to_string(image)

    def extract_with_confidence(self, image_path: str) -> List[Dict[str, Any]]:
        image = Image.open(image_path)
        data = pytesseract.image_to_data(image, output_type=pytesseract.Output.DICT)
        results = []
        for i in range(len(data["text"])):
            if data["text"][i].strip():
                results.append({
                    "text": data["text"][i],
                    "confidence": data["conf"][i],
                    "bbox": {
                        "x": data["left"][i],
                        "y": data["top"][i],
                        "w": data["width"][i],
                        "h": data["height"][i],
                    },
                })
        return results

    def extract_tables(self, image_path: str) -> List[List[str]]:
        image = Image.open(image_path)
        text = pytesseract.image_to_string(image)
        rows = [row.split("\t") for row in text.strip().split("\n") if row.strip()]
        return rows


class ChartParser:
    """Parse charts and diagrams using GPT-4 Vision."""

    def __init__(self, model: str = "gpt-4o"):
        self.client = AsyncOpenAI()
        self.model = model

    async def parse_chart(self, image_path: str) -> Dict[str, Any]:
        with open(image_path, "rb") as f:
            base64_image = base64.b64encode(f.read()).decode("utf-8")
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": """Analyze this chart/graph and extract:
1. Chart type (bar, line, pie, scatter, etc.)
2. Title
3. Axes labels
4. Data points (as many as visible)
5. Key insights/trends

Return as JSON.""",
                        },
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{base64_image}",
                                "detail": "high",
                            },
                        },
                    ],
                }
            ],
            max_tokens=2000,
        )
        try:
            return json.loads(response.choices[0].message.content)
        except json.JSONDecodeError:
            return {"chart_type": "unknown", "raw": response.choices[0].message.content}

class VisualAgent:
    """Complete visual analysis agent orchestrating vision, OCR, and chart parsing."""

    def __init__(self, model: str = "gpt-4o"):
        self.vision = GPT4VisionClient(model)
        self.ocr = OCRExtractor()
        self.chart_parser = ChartParser(model)

    async def analyze(self, image_path: str, task: str = "describe") -> Dict[str, Any]:
        if task == "describe":
            description = await self.vision.analyze_image(image_path, "Describe this image in detail.")
            return {"type": "description", "content": description}
        elif task == "ocr":
            text = self.ocr.extract_text(image_path)
            return {"type": "ocr", "content": text}
        elif task == "chart":
            chart_data = await self.chart_parser.parse_chart(image_path)
            return {"type": "chart", "content": chart_data}
        elif task == "extract":
            schema = '{"text": [], "numbers": [], "labels": []}'
            data = await self.vision.extract_structured_data(image_path, schema)
            return {"type": "extraction", "content": data}
        else:
            answer = await self.vision.analyze_image(image_path, task)
            return {"type": "qa", "content": answer}

    async def process_document_images(self, image_paths: List[str]) -> Dict[str, Any]:
        all_text = []
        all_descriptions = []
        for path in image_paths:
            text = self.ocr.extract_text(path)
            all_text.append(text)
            desc = await self.vision.analyze_image(path, "Summarize this document page.")
            all_descriptions.append(desc)
        return {
            "pages": len(image_paths),
            "extracted_text": "\n\n".join(all_text),
            "summaries": all_descriptions,
        }