mirror-epstein-docs/process_images.py

#!/usr/bin/env python3
"""
Image processing script for OCR and entity extraction using OpenAI-compatible API.
Processes images from Downloads folder and extracts structured data.
"""

import os
import json
import base64
from pathlib import Path
from typing import Dict, List, Optional
import concurrent.futures
from dataclasses import dataclass, asdict
from openai import OpenAI
from tqdm import tqdm
import argparse
from dotenv import load_dotenv


@dataclass
class ProcessingResult:
    """Structure for processing results"""
    filename: str
    success: bool
    data: Optional[Dict] = None
    error: Optional[str] = None


class ImageProcessor:
    """Process images using OpenAI-compatible vision API"""

    def __init__(self, api_url: str, api_key: str, model: str = "gpt-4o", index_file: str = "processing_index.json", downloads_dir: Optional[str] = None):
        self.client = OpenAI(api_key=api_key, base_url=api_url)
        self.model = model
        self.downloads_dir = Path(downloads_dir) if downloads_dir else Path.home() / "Downloads"
        self.index_file = index_file
        self.processed_files = self.load_index()

    def load_index(self) -> set:
        """Load the index of already processed files"""
        if os.path.exists(self.index_file):
            try:
                with open(self.index_file, 'r') as f:
                    data = json.load(f)
                    return set(data.get('processed_files', []))
            except Exception as e:
                print(f"⚠️  Warning: Could not load index file: {e}")
                return set()
        return set()

    def save_index(self):
        """Save the current index of processed files"""
        with open(self.index_file, 'w') as f:
            json.dump({
                'processed_files': sorted(list(self.processed_files)),
                'last_updated': str(Path.cwd())
            }, f, indent=2)

    def mark_processed(self, filename: str):
        """Mark a file as processed and update index"""
        self.processed_files.add(filename)
        self.save_index()

    def get_image_files(self) -> List[Path]:
        """Get all image files from Downloads folder (recursively)"""
        image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp'}
        image_files = []

        for ext in image_extensions:
            image_files.extend(self.downloads_dir.glob(f'**/*{ext}'))
            image_files.extend(self.downloads_dir.glob(f'**/*{ext.upper()}'))

        return sorted(image_files)

    def get_relative_path(self, file_path: Path) -> str:
        """Get relative path from downloads directory for unique indexing"""
        try:
            return str(file_path.relative_to(self.downloads_dir))
        except ValueError:
            # If file is not relative to downloads_dir, use full path
            return str(file_path)

    def get_unprocessed_files(self) -> List[Path]:
        """Get only files that haven't been processed yet"""
        all_files = self.get_image_files()
        unprocessed = [f for f in all_files if self.get_relative_path(f) not in self.processed_files]
        return unprocessed

    def encode_image(self, image_path: Path) -> str:
        """Encode image to base64"""
        with open(image_path, 'rb') as f:
            return base64.b64encode(f.read()).decode('utf-8')

    def get_system_prompt(self) -> str:
        """Get the system prompt for structured extraction"""
        return """You are an expert OCR and document analysis system.
Extract ALL text from the image in READING ORDER to create a digital twin of the document.

IMPORTANT: Transcribe text exactly as it appears on the page, from top to bottom, left to right, including:
- All printed text
- All handwritten text (inline where it appears)
- Stamps and annotations (inline where they appear)
- Signatures (note location)

Preserve the natural reading flow. Mix printed and handwritten text together in the order they appear.

Return ONLY valid JSON in this exact structure:
{
  "document_metadata": {
    "page_number": "string or null",
    "document_number": "string or null",
    "date": "string or null",
    "document_type": "string or null",
    "has_handwriting": true/false,
    "has_stamps": true/false
  },
  "full_text": "Complete text transcription in reading order. Include ALL text - printed, handwritten, stamps, etc. - exactly as it appears from top to bottom.",
  "text_blocks": [
    {
      "type": "printed|handwritten|stamp|signature|other",
      "content": "text content",
      "position": "top|middle|bottom|header|footer|margin"
    }
  ],
  "entities": {
    "people": ["list of person names"],
    "organizations": ["list of organizations"],
    "locations": ["list of locations"],
    "dates": ["list of dates found"],
    "reference_numbers": ["list of any reference/ID numbers"]
  },
  "additional_notes": "Any observations about document quality, redactions, damage, etc."
}"""

    def process_image(self, image_path: Path) -> ProcessingResult:
        """Process a single image through the API"""
        try:
            # Encode image
            base64_image = self.encode_image(image_path)

            # Make API call using OpenAI client
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {
                        "role": "system",
                        "content": self.get_system_prompt()
                    },
                    {
                        "role": "user",
                        "content": [
                            {
                                "type": "text",
                                "text": "Extract all text and entities from this image. Return only valid JSON."
                            },
                            {
                                "type": "image_url",
                                "image_url": {
                                    "url": f"data:image/jpeg;base64,{base64_image}"
                                }
                            }
                        ]
                    }
                ],
                max_tokens=4096,
                temperature=0.1
            )

            # Parse response
            content = response.choices[0].message.content

            # Parse JSON from content (strip markdown if present)
            content = content.strip()
            if content.startswith('```json'):
                content = content[7:]
            if content.startswith('```'):
                content = content[3:]
            if content.endswith('```'):
                content = content[:-3]
            content = content.strip()

            extracted_data = json.loads(content)

            return ProcessingResult(
                filename=self.get_relative_path(image_path),
                success=True,
                data=extracted_data
            )

        except Exception as e:
            return ProcessingResult(
                filename=self.get_relative_path(image_path),
                success=False,
                error=str(e)
            )

    def process_all(self, max_workers: int = 5, limit: Optional[int] = None, resume: bool = True) -> List[ProcessingResult]:
        """Process all images with parallel processing"""
        if resume:
            image_files = self.get_unprocessed_files()
            total_files = len(self.get_image_files())
            already_processed = len(self.processed_files)
            print(f"Found {total_files} total image files")
            print(f"Already processed: {already_processed}")
            print(f"Remaining to process: {len(image_files)}")
        else:
            image_files = self.get_image_files()
            print(f"Found {len(image_files)} image files to process")

        if limit:
            image_files = image_files[:limit]
            print(f"Limited to {limit} files for this run")

        if not image_files:
            print("No files to process!")
            return []

        results = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {executor.submit(self.process_image, img): img for img in image_files}

            with tqdm(total=len(image_files), desc="Processing images") as pbar:
                for future in concurrent.futures.as_completed(futures):
                    result = future.result()
                    results.append(result)

                    # Save individual result to file
                    if result.success:
                        self.save_individual_result(result)

                    # Mark as processed regardless of success/failure
                    self.mark_processed(result.filename)

                    pbar.update(1)

                    if not result.success:
                        tqdm.write(f"❌ Failed: {result.filename} - {result.error}")
                    else:
                        tqdm.write(f"✅ Processed: {result.filename}")

        return results

    def save_individual_result(self, result: ProcessingResult):
        """Save individual result to ./results/folder/imagename.json"""
        # Create output path mirroring the source structure
        result_path = Path("./results") / result.filename
        result_path = result_path.with_suffix('.json')

        # Create parent directories
        result_path.parent.mkdir(parents=True, exist_ok=True)

        # Save the extracted data
        with open(result_path, 'w', encoding='utf-8') as f:
            json.dump(result.data, f, indent=2, ensure_ascii=False)

    def save_results(self, results: List[ProcessingResult], output_file: str = "processed_results.json"):
        """Save summary results to JSON file"""
        output_data = {
            "total_processed": len(results),
            "successful": sum(1 for r in results if r.success),
            "failed": sum(1 for r in results if not r.success),
            "results": [asdict(r) for r in results]
        }

        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(output_data, f, indent=2, ensure_ascii=False)

        print(f"\n✅ Summary saved to {output_file}")
        print(f"   Individual results saved to ./results/")
        print(f"   Successful: {output_data['successful']}")
        print(f"   Failed: {output_data['failed']}")


def main():
    # Load environment variables
    load_dotenv()

    parser = argparse.ArgumentParser(description="Process images with OCR and entity extraction")
    parser.add_argument("--api-url", help="OpenAI-compatible API base URL (default: from .env or OPENAI_API_URL)")
    parser.add_argument("--api-key", help="API key (default: from .env or OPENAI_API_KEY)")
    parser.add_argument("--model", help="Model name (default: from .env, OPENAI_MODEL, or meta-llama/Llama-4-Maverick-17B-128E-Instruct)")
    parser.add_argument("--workers", type=int, default=5, help="Number of parallel workers (default: 5)")
    parser.add_argument("--limit", type=int, help="Limit number of images to process (for testing)")
    parser.add_argument("--output", default="processed_results.json", help="Output JSON file")
    parser.add_argument("--index", default="processing_index.json", help="Index file to track processed files")
    parser.add_argument("--downloads-dir", default="./downloads", help="Directory containing images (default: ./downloads)")
    parser.add_argument("--no-resume", action="store_true", help="Process all files, ignoring index")

    args = parser.parse_args()

    # Get values from args or environment variables
    api_url = args.api_url or os.getenv("OPENAI_API_URL", "http://...")
    api_key = args.api_key or os.getenv("OPENAI_API_KEY", "abcd1234")
    model = args.model or os.getenv("OPENAI_MODEL", "meta-llama/Llama-4-Maverick-17B-128E-Instruct")

    processor = ImageProcessor(
        api_url=api_url,
        api_key=api_key,
        model=model,
        index_file=args.index,
        downloads_dir=args.downloads_dir
    )

    results = processor.process_all(
        max_workers=args.workers,
        limit=args.limit,
        resume=not args.no_resume
    )

    processor.save_results(results, args.output)


if __name__ == "__main__":
    main()