# Create a new project directory
mkdir ai-document-processor
cd ai-document-processor

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install core dependencies
pip install requests python-dotenv tqdm pypdf2 pillow
pip install pandas numpy  # For data handling
pip install tenacity  # For retry logic

DEEPSEEK_API_KEY=your_api_key_here
DEEPSEEK_BASE_URL=https://api.deepseek.com
BATCH_SIZE=50
MAX_RETRIES=5
RATE_LIMIT_PER_MINUTE=100
OUTPUT_DIR=./processed_results

import os
import json
import base64
from pathlib import Path
from typing import List, Dict, Any, Optional
import PyPDF2
from PIL import Image
import io

class DocumentProcessor:
    """Handles loading and preprocessing of various document types"""
    
    def __init__(self, input_dir: str):
        self.input_dir = Path(input_dir)
        self.supported_extensions = {'.pdf', '.txt', '.png', '.jpg', '.jpeg'}
    
    def load_documents(self) -> List[Dict[str, Any]]:
        """Load all documents from the input directory"""
        documents = []
        
        for file_path in self.input_dir.rglob('*'):
            if file_path.suffix.lower() in self.supported_extensions:
                try:
                    content = self._read_document(file_path)
                    documents.append({
                        'id': str(file_path.relative_to(self.input_dir)),
                        'path': str(file_path),
                        'content': content,
                        'size': len(content),
                        'extension': file_path.suffix.lower()
                    })
                except Exception as e:
                    print(f"Error loading {file_path}: {e}")
        
        return documents
    
    def _read_document(self, file_path: Path) -> str:
        """Read document content based on file type"""
        extension = file_path.suffix.lower()
        
        if extension == '.pdf':
            return self._extract_text_from_pdf(file_path)
        elif extension in ['.png', '.jpg', '.jpeg']:
            return self._encode_image(file_path)
        elif extension == '.txt':
            return file_path.read_text(encoding='utf-8')
        else:
            raise ValueError(f"Unsupported file type: {extension}")
    
    def _extract_text_from_pdf(self, file_path: Path) -> str:
        """Extract text from PDF files"""
        text = []
        with open(file_path, 'rb') as file:
            pdf_reader = PyPDF2.PdfReader(file)
            for page_num in range(len(pdf_reader.pages)):
                page = pdf_reader.pages[page_num]
                text.append(page.extract_text())
        return '\n'.join(text)
    
    def _encode_image(self, file_path: Path) -> str:
        """Encode image to base64 for API consumption"""
        with open(file_path, 'rb') as image_file:
            encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
        return encoded_string
    
    def estimate_tokens(self, text: str) -> int:
        """Simple token estimation (approximate)"""
        # Rough estimate: 1 token ≈ 0.75 words
        words = len(text.split())
        return int(words * 1.33)

import requests
import time
from typing import Dict, Any, List
import json
from tenacity import retry, stop_after_attempt, wait_exponential
from dataclasses import dataclass
from datetime import datetime

@dataclass
class CostTracker:
    """Track API usage and costs in real-time"""
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    total_requests: int = 0
    total_cost: float = 0.0
    
    # DeepSeek pricing (approximate USD)
    INPUT_COST_PER_MILLION = 0.019  # $0.019 per 1M input tokens
    OUTPUT_COST_PER_MILLION = 0.039  # $0.039 per 1M output tokens
    
    def add_usage(self, input_tokens: int, output_tokens: int):
        """Add token usage and calculate incremental cost"""
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens
        self.total_requests += 1
        
        input_cost = (input_tokens / 1_000_000) * self.INPUT_COST_PER_MILLION
        output_cost = (output_tokens / 1_000_000) * self.OUTPUT_COST_PER_MILLION
        self.total_cost += input_cost + output_cost
    
    def get_summary(self) -> Dict[str, Any]:
        """Get current cost and usage summary"""
        return {
            'total_input_tokens': self.total_input_tokens,
            'total_output_tokens': self.total_output_tokens,
            'total_requests': self.total_requests,
            'estimated_cost_usd': round(self.total_cost, 4),
            'cost_per_document': round(self.total_cost / max(self.total_requests, 1), 6),
            'timestamp': datetime.now().isoformat()
        }

class DeepSeekClient:
    """Client for DeepSeek API with built-in cost tracking and retry logic"""
    
    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com"):
        self.api_key = api_key
        self.base_url = base_url
        self.cost_tracker = CostTracker()
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        })
    
    @retry(
        stop=stop_after_attempt(5),
        wait=wait_exponential(multiplier=1, min=4, max=60)
    )
    def process_document(self, 
                        document_content: str,
                        extraction_schema: Dict[str, Any],
                        model: str = "deepseek-chat") -> Dict[str, Any]:
        """
        Process a single document with structured extraction
        
        Args:
            document_content: Text or base64 encoded image
            extraction_schema: JSON schema for structured output
            model: DeepSeek model to use
        
        Returns:
            Structured extraction results
        """
        
        # Optimized system prompt for minimal token usage
        system_prompt = f"""Extract information from this document according to the schema.
        Return ONLY valid JSON matching this structure: {json.dumps(extraction_schema)}
        Be concise. Use null for missing values."""
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": document_content}
        ]
        
        payload = {
            "model": model,
            "messages": messages,
            "response_format": {"type": "json_object"},
            "max_tokens": 500,  # Limit output to control costs
            "temperature": 0.1  # Low temperature for consistent formatting
        }
        
        try:
            response = self.session.post(
                f"{self.base_url}/chat/completions",
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            
            result = response.json()
            
            # Extract token usage and track costs
            usage = result.get('usage', {})
            input_tokens = usage.get('prompt_tokens', 0)
            output_tokens = usage.get('completion_tokens', 0)
            
            self.cost_tracker.add_usage(input_tokens, output_tokens)
            
            # Parse the response content
            content = result['choices'][0]['message']['content']
            return json.loads(content)
            
        except requests.exceptions.RequestException as e:
            print(f"API request failed: {e}")
            raise
        except json.JSONDecodeError as e:
            print(f"Failed to parse JSON response: {e}")
            raise
    
    def batch_process(self, 
                     documents: List[Dict[str, Any]],
                     schema: Dict[str, Any],
                     batch_size: int = 10,
                     delay_between_batches: float = 1.0) -> List[Dict[str, Any]]:
        """
        Process documents in batches for efficiency
        
        Args:
            documents: List of document dictionaries
            schema: Extraction schema
            batch_size: Number of documents per batch
            delay_between_batches: Delay to respect rate limits
        
        Returns:
            List of processed results
        """
        results = []
        
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i + batch_size]
            batch_results = []
            
            print(f"Processing batch {i//batch_size + 1}/{(len(documents)-1)//batch_size + 1}")
            
            for doc in batch:
                try:
                    result = self.process_document(doc['content'], schema)
                    result['document_id'] = doc['id']
                    result['processing_time'] = datetime.now().isoformat()
                    batch_results.append(result)
                    
                    # Print cost update every 100 documents
                    if len(results) % 100 == 0:
                        summary = self.cost_tracker.get_summary()
                        print(f"Progress: {len(results)} documents, Cost: ${summary['estimated_cost_usd']}")
                    
                except Exception as e:
                    print(f"Failed to process document {doc['id']}: {e}")
                    # Store error for later retry
                    batch_results.append({
                        'document_id': doc['id'],
                        'error': str(e),
                        'processed': False
                    })
            
            results.extend(batch_results)
            
            # Respect rate limits
            if i + batch_size < len(documents):
                time.sleep(delay_between_batches)
        
        return results

import concurrent.futures
from tqdm import tqdm
import pandas as pd
import logging

class LargeScaleProcessor:
    """Main orchestrator for large-scale document processing"""
    
    def __init__(self, 
                 api_key: str,
                 input_dir: str,
                 output_dir: str,
                 max_workers: int = 10):
        self.api_key = api_key
        self.input_dir = input_dir
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.max_workers = max_workers
        
        # Initialize components
        self.doc_processor = DocumentProcessor(input_dir)
        self.api_client = DeepSeekClient(api_key)
        
        # Setup logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(self.output_dir / 'processing.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def define_extraction_schema(self, doc_type: str) -> Dict[str, Any]:
        """Define extraction schema based on document type"""
        
        # Example schema for invoices
        if doc_type == 'invoice':
            return {
                "type": "object",
                "properties": {
                    "invoice_number": {"type": "string"},
                    "date": {"type": "string"},
                    "vendor_name": {"type": "string"},
                    "total_amount": {"type": "number"},
                    "tax_amount": {"type": "number"},
                    "line_items": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "description": {"type": "string"},
                                "quantity": {"type": "number"},
                                "unit_price": {"type": "number"},
                                "total": {"type": "number"}
                            }
                        }
                    }
                },
                "required": ["invoice_number", "date", "total_amount"]
            }
        
        # Default schema for general documents
        return {
            "type": "object",
            "properties": {
                "summary": {"type": "string"},
                "key_points": {
                    "type": "array",
                    "items": {"type": "string"}
                },
                "entities": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "type": {"type": "string"},
                            "context": {"type": "string"}
                        }
                    }
                },
                "sentiment": {"type": "string", "enum": ["positive", "neutral", "negative"]}
            }
        }
    
    def process_single_document(self, document: Dict[str, Any], schema: Dict[str, Any]) -> Dict[str, Any]:
        """Process a single document with error handling"""
        try:
            result = self.api_client.process_document(document['content'], schema)
            result['document_id'] = document['id']
            result['processing_success'] = True
            return result
        except Exception as e:
            self.logger.error(f"Failed to process {document['id']}: {e}")
            return {
                'document_id': document['id'],
                'processing_success': False,
                'error': str(e),
                'content_preview': document['content'][:500] if len(document['content']) > 500 else document['content']
            }
    
    def run_parallel_processing(self, 
                               doc_type: str = "general",
                               batch_size: int = 1000) -> None:
        """
        Main processing method with parallel execution
        
        Args:
            doc_type: Type of documents being processed
            batch_size: Number of documents to process in memory at once
        """
        
        # Load documents
        self.logger.info("Loading documents...")
        all_documents = self.doc_processor.load_documents()
        self.logger.info(f"Loaded {len(all_documents)}