main.py

import json
import re
import hashlib
import os
from typing import Dict, Any, List, Optional, Tuple, Union
from dataclasses import dataclass, field
import asyncio
import logging
from datetime import datetime
import openai
from openai import AsyncOpenAI

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ComplexityMetrics:
    max_depth: int
    total_fields: int
    enum_count: int
    required_fields: int
    nested_objects: int
    
    @property
    def complexity_tier(self) -> int:
        if self.max_depth <= 2 and self.total_fields <= 20:
            return 1
        elif self.max_depth <= 4 and self.total_fields <= 100:
            return 2
        else:
            return 3

@dataclass
class ExtractionStage:
    name: str
    fields: List[str]
    schema_subset: Dict[str, Any]
    complexity: int
    dependencies: List[str] = field(default_factory=list)
    estimated_tokens: int = 0

@dataclass
class ExtractionPlan:
    stages: List[ExtractionStage]
    estimated_cost: float
    estimated_time: float
    model_assignments: Dict[str, str]
    parallelizable_stages: List[str] = field(default_factory=list)

@dataclass
class ExtractionResult:
    data: Dict[str, Any]
    confidence_scores: Dict[str, float]
    stage_results: List[Dict[str, Any]] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    processing_time: float = 0.0

@dataclass
class QualityReport:
    overall_confidence: float
    field_scores: Dict[str, float]
    review_flags: List[str]
    schema_compliance: float
    consistency_score: float
    recommended_review_time: int = 0

class OpenAIClient:
    def __init__(self, model_name: str, api_key: str):
        self.model_name = model_name
        self.client = AsyncOpenAI(api_key=api_key)
        self.cost_per_token = {
            "gpt-4o-mini": 0.00015,
            "gpt-4o": 0.005,
            "gpt-4-turbo": 0.003
        }
    
    async def complete(self, prompt: str, max_tokens: int = 4000) -> Tuple[str, float]:
        try:
            response = await self.client.chat.completions.create(
                model=self.model_name,
                messages=[
                    {"role": "system", "content": "You are a precise data extraction specialist. Extract data according to the provided schema and output only valid JSON."},
                    {"role": "user", "content": prompt}
                ],
                max_tokens=max_tokens,
                temperature=0.1,
                top_p=0.9
            )
            
            content = response.choices[0].message.content
            confidence = 0.9 if "gpt-4o" in self.model_name else 0.8
            
            if content and len(content.strip()) > 10:
                confidence += 0.05
            
            return content, confidence
            
        except Exception as e:
            logger.error(f"OpenAI API error: {e}")
            return '{"error": "API call failed", "details": "' + str(e) + '"}', 0.1

class SchemaAnalyzer:
    def analyze_complexity(self, schema: Dict[str, Any]) -> ComplexityMetrics:
        def count_depth(obj: Any, current_depth: int = 0) -> int:
            if not isinstance(obj, dict):
                return current_depth
            
            max_child_depth = current_depth
            for value in obj.values():
                if isinstance(value, dict):
                    if 'properties' in value:
                        child_depth = count_depth(value['properties'], current_depth + 1)
                    else:
                        child_depth = count_depth(value, current_depth + 1)
                    max_child_depth = max(max_child_depth, child_depth)
            return max_child_depth
        
        def count_fields(obj: Any) -> Tuple[int, int, int]:
            if not isinstance(obj, dict):
                return 0, 0, 0
            
            total, enums, objects = 0, 0, 0
            
            for key, value in obj.items():
                if key == 'properties' and isinstance(value, dict):
                    for prop_name, prop_def in value.items():
                        total += 1
                        if isinstance(prop_def, dict):
                            if 'enum' in prop_def:
                                enums += 1
                            if prop_def.get('type') == 'object':
                                objects += 1
                            nested_total, nested_enums, nested_objects = count_fields(prop_def)
                            total += nested_total
                            enums += nested_enums
                            objects += nested_objects
                elif isinstance(value, dict):
                    nested_total, nested_enums, nested_objects = count_fields(value)
                    total += nested_total
                    enums += nested_enums
                    objects += nested_objects
            
            return total, enums, objects
        
        max_depth = count_depth(schema.get('properties', {}))
        total_fields, enum_count, nested_objects = count_fields(schema)
        required_fields = len(schema.get('required', []))
        
        return ComplexityMetrics(
            max_depth=max_depth,
            total_fields=total_fields,
            enum_count=enum_count,
            required_fields=required_fields,
            nested_objects=nested_objects
        )
    
    def create_extraction_plan(self, schema: Dict[str, Any], complexity: ComplexityMetrics) -> ExtractionPlan:
        return self._create_single_pass_plan(schema)
    
    def _create_single_pass_plan(self, schema: Dict[str, Any]) -> ExtractionPlan:
        stages = [ExtractionStage(
            name="complete_extraction",
            fields=list(schema.get('properties', {}).keys()),
            schema_subset=schema,
            complexity=2,
            estimated_tokens=4000
        )]
        
        return ExtractionPlan(
            stages=stages,
            estimated_cost=0.15,
            estimated_time=15.0,
            model_assignments={"complete_extraction": "gpt-4o"}
        )
    
    def _create_simple_plan(self, schema: Dict[str, Any]) -> ExtractionPlan:
        stages = [ExtractionStage(
            name="complete_extraction",
            fields=list(schema.get('properties', {}).keys()),
            schema_subset=schema,
            complexity=1,
            estimated_tokens=2000
        )]
        
        return ExtractionPlan(
            stages=stages,
            estimated_cost=0.02,
            estimated_time=5.0,
            model_assignments={"complete_extraction": "gpt-4o"}
        )
    
    def _create_medium_plan(self, schema: Dict[str, Any]) -> ExtractionPlan:
        properties = schema.get('properties', {})
        simple_fields = []
        complex_fields = []
        
        for field_name, field_def in properties.items():
            if isinstance(field_def, dict) and field_def.get('type') in ['object', 'array']:
                complex_fields.append(field_name)
            else:
                simple_fields.append(field_name)
        
        stages = []
        if simple_fields:
            stages.append(ExtractionStage(
                name="simple_fields",
                fields=simple_fields,
                schema_subset=self._create_subset_schema(schema, simple_fields),
                complexity=1,
                estimated_tokens=1500
            ))
        
        if complex_fields:
            stages.append(ExtractionStage(
                name="complex_fields",
                fields=complex_fields,
                schema_subset=self._create_subset_schema(schema, complex_fields),
                complexity=2,
                dependencies=["simple_fields"] if simple_fields else [],
                estimated_tokens=3000
            ))
        
        return ExtractionPlan(
            stages=stages,
            estimated_cost=0.15,
            estimated_time=25.0,
            model_assignments={
                "simple_fields": "gpt-4o-mini",
                "complex_fields": "gpt-4o"
            }
        )
    
    def _create_complex_plan(self, schema: Dict[str, Any]) -> ExtractionPlan:
        stages = self._create_hierarchical_stages(schema)
        
        model_assignments = {
            stage.name: "gpt-4o" if stage.complexity > 1 else "gpt-4o-mini"
            for stage in stages
        }
        
        estimated_cost = len(stages) * 0.10
        estimated_time = len(stages) * 15.0
        
        return ExtractionPlan(
            stages=stages,
            estimated_cost=min(estimated_cost, 2.0),
            estimated_time=min(estimated_time, 120.0),
            model_assignments=model_assignments
        )
    
    def _create_hierarchical_stages(self, schema: Dict[str, Any]) -> List[ExtractionStage]:
        stages = []
        properties = schema.get('properties', {})
        
        simple_fields = [
            field_name for field_name, field_def in properties.items()
            if isinstance(field_def, dict) and field_def.get('type') in ['string', 'number', 'integer', 'boolean']
            and 'enum' not in field_def
        ]
        
        if simple_fields:
            stages.append(ExtractionStage(
                name="primitive_fields",
                fields=simple_fields,
                schema_subset=self._create_subset_schema(schema, simple_fields),
                complexity=1,
                estimated_tokens=1000
            ))
        
        enum_fields = [
            field_name for field_name, field_def in properties.items()
            if isinstance(field_def, dict) and 'enum' in field_def
        ]
        
        if enum_fields:
            stages.append(ExtractionStage(
                name="enum_fields",
                fields=enum_fields,
                schema_subset=self._create_subset_schema(schema, enum_fields),
                complexity=1,
                dependencies=["primitive_fields"] if simple_fields else [],
                estimated_tokens=1500
            ))
        
        array_fields = [
            field_name for field_name, field_def in properties.items()
            if isinstance(field_def, dict) and field_def.get('type') == 'array'
        ]
        
        if array_fields:
            stages.append(ExtractionStage(
                name="array_fields",
                fields=array_fields,
                schema_subset=self._create_subset_schema(schema, array_fields),
                complexity=2,
                dependencies=["primitive_fields", "enum_fields"],
                estimated_tokens=2500
            ))
        
        object_fields = [
            field_name for field_name, field_def in properties.items()
            if isinstance(field_def, dict) and field_def.get('type') == 'object'
        ]
        
        if object_fields:
            stages.append(ExtractionStage(
                name="object_fields",
                fields=object_fields,
                schema_subset=self._create_subset_schema(schema, object_fields),
                complexity=3,
                dependencies=["primitive_fields", "enum_fields", "array_fields"],
                estimated_tokens=4000
            ))
        
        return [stage for stage in stages if stage.fields]
    
    def _create_subset_schema(self, full_schema: Dict[str, Any], fields: List[str]) -> Dict[str, Any]:
        properties = full_schema.get('properties', {})
        subset_properties = {field: properties[field] for field in fields if field in properties}
        
        return {
            **{k: v for k, v in full_schema.items() if k != 'properties'},
            'properties': subset_properties
        }

class DocumentProcessor:
    def __init__(self, max_chunk_size: int = 100000):
        self.max_chunk_size = max_chunk_size
    
    def process_document(self, content: str, schema: Dict[str, Any]) -> List[str]:
        if len(content) <= self.max_chunk_size:
            return [content]
        
        logger.info(f"Document size {len(content)} exceeds chunk limit, creating semantic chunks")
        return self._semantic_chunking(content, schema)
    
    def _semantic_chunking(self, content: str, schema: Dict[str, Any]) -> List[str]:
        paragraphs = content.split('\n\n')
        chunks = []
        current_chunk = ""
        overlap_size = 1000
        
        for para in paragraphs:
            if len(current_chunk) + len(para) > self.max_chunk_size:
                if current_chunk:
                    chunks.append(current_chunk)
                    current_chunk = current_chunk[-overlap_size:] + "\n\n" + para
                else:
                    current_chunk = para
            else:
                current_chunk += "\n\n" + para if current_chunk else para
        
        if current_chunk:
            chunks.append(current_chunk)
        
        logger.info(f"Created {len(chunks)} semantic chunks")
        return chunks

class ExtractionEngine:
    def __init__(self, api_key: str):
        self.models = {
            "gpt-4o-mini": OpenAIClient("gpt-4o-mini", api_key),
            "gpt-4o": OpenAIClient("gpt-4o", api_key),
        }
    
    async def extract(self, content: str, plan: ExtractionPlan, schema: Dict[str, Any]) -> ExtractionResult:
        start_time = asyncio.get_event_loop().time()
        results = {}
        confidence_scores = {}
        stage_results = []
        
        logger.info(f"Starting extraction with {len(plan.stages)} stages")
        
        for i, stage in enumerate(plan.stages):
            logger.info(f"Executing stage {i+1}/{len(plan.stages)}: {stage.name}")
            
            if not self._dependencies_satisfied(stage.dependencies, results):
                logger.warning(f"Dependencies not satisfied for stage {stage.name}, skipping")
                continue
            
            context = self._build_context(content, results, stage)
            model_name = plan.model_assignments.get(stage.name, "gpt-4o")
            model = self.models[model_name]
            
            prompt = self._create_extraction_prompt(context, stage.schema_subset, results)
            
            response, confidence = await model.complete(prompt, max_tokens=4000)
            stage_data = self._parse_response(response, stage.fields)
            
            results.update(stage_data)
            for field in stage.fields:
                confidence_scores[field] = confidence * (0.9 if field in stage_data else 0.3)
            
            stage_results.append({
                "stage": stage.name,
                "extracted_fields": list(stage_data.keys()),
                "confidence": confidence,
                "model": model_name,
                "processing_time": 0.5
            })
        
        processing_time = asyncio.get_event_loop().time() - start_time
        
        return ExtractionResult(
            data=results,
            confidence_scores=confidence_scores,
            stage_results=stage_results,
            metadata={
                "total_stages": len(plan.stages),
                "estimated_cost": plan.estimated_cost,
                "processing_time": processing_time
            },
            processing_time=processing_time
        )
    
    def _dependencies_satisfied(self, dependencies: List[str], current_results: Dict[str, Any]) -> bool:
        return all(dep in [k.split('.')[0] for k in current_results.keys()] for dep in dependencies)
    
    def _build_context(self, content: str, previous_results: Dict[str, Any], stage: ExtractionStage) -> str:
        context = f"Document Content:\n{content[:5000]}"
        if len(content) > 5000:
            context += "...[truncated]"
        
        if previous_results:
            context += f"\n\nPreviously Extracted Data:\n{json.dumps(previous_results, indent=2)[:1000]}"
        
        return context
    
    def _create_extraction_prompt(self, context: str, schema: Dict[str, Any], previous_results: Dict[str, Any]) -> str:
        schema_properties = schema.get('properties', {})
        required_fields = schema.get('required', [])
        
        field_descriptions = []
        for field_name, field_def in schema_properties.items():
            if isinstance(field_def, dict):
                field_type = field_def.get('type', 'string')
                is_required = field_name in required_fields
                status = "REQUIRED" if is_required else "optional"
                field_descriptions.append(f"- {field_name} ({field_type}) [{status}]")
        
        previous_context = ""
        if previous_results:
            previous_context = f"\n\nPreviously extracted data:\n{json.dumps(previous_results, indent=2)}"
        
        return f"""Extract ALL specified fields from the document content according to the JSON schema.

DOCUMENT CONTENT:
{context[:4000]}

REQUIRED OUTPUT FIELDS:
{chr(10).join(field_descriptions)}

SCHEMA STRUCTURE:
{json.dumps(schema, indent=2)}{previous_context}

CRITICAL INSTRUCTIONS:
1. Extract ALL fields specified in the schema properties
2. For arrays, extract ALL items found in the content
3. For objects, extract ALL nested properties
4. Use null only if data truly cannot be found
5. Maintain exact schema structure and types
6. Output ONLY valid JSON, no explanations

JSON OUTPUT:"""
    
    def _parse_response(self, response: str, expected_fields: List[str]) -> Dict[str, Any]:
        try:
            cleaned_response = response.strip()
            
            if not cleaned_response.startswith('{'):
                json_start = cleaned_response.find('{')
                if json_start != -1:
                    cleaned_response = cleaned_response[json_start:]
            
            if not cleaned_response.endswith('}'):
                json_end = cleaned_response.rfind('}')
                if json_end != -1:
                    cleaned_response = cleaned_response[:json_end + 1]
            
            data = json.loads(cleaned_response)
            
            if isinstance(data, dict):
                return data
            else:
                logger.warning("Response is not a dictionary")
                return {}
                
        except json.JSONDecodeError as e:
            logger.warning(f"JSON decode error: {e}")
            
            try:
                import re
                json_pattern = r'\{(?:[^{}]|{(?:[^{}]|{[^{}]*})*})*\}'
                matches = re.findall(json_pattern, response, re.DOTALL)
                
                for match in matches:
                    try:
                        data = json.loads(match)
                        if isinstance(data, dict) and data:
                            return data
                    except:
                        continue
                        
            except Exception as e:
                logger.warning(f"Regex parsing failed: {e}")
            
            logger.error("All JSON parsing attempts failed")
            return {}

class QualityAssessor:
    def assess_extraction(self, result: ExtractionResult, schema: Dict[str, Any]) -> QualityReport:
        schema_compliance = self._validate_against_schema(result.data, schema)
        field_scores = result.confidence_scores.copy()
        consistency_score = self._check_consistency(result.data)
        
        required_fields = schema.get('required', [])
        total_expected_fields = len(schema.get('properties', {}))
        extracted_fields = len([k for k, v in result.data.items() if v is not None])
        
        completeness_score = extracted_fields / total_expected_fields if total_expected_fields > 0 else 0
        
        if field_scores:
            avg_field_confidence = sum(field_scores.values()) / len(field_scores)
        else:
            avg_field_confidence = 0
        
        overall_confidence = completeness_score * 0.6 + schema_compliance * 0.3 + consistency_score * 0.1
        overall_confidence = min(overall_confidence, 1.0)
        
        review_flags = self._generate_review_flags(field_scores, schema_compliance, overall_confidence, required_fields, result.data, total_expected_fields, extracted_fields)
        review_time = self._estimate_review_time(review_flags, field_scores)
        
        return QualityReport(
            overall_confidence=overall_confidence,
            field_scores=field_scores,
            review_flags=review_flags,
            schema_compliance=schema_compliance,
            consistency_score=consistency_score,
            recommended_review_time=review_time
        )
    
    def _validate_against_schema(self, data: Dict[str, Any], schema: Dict[str, Any]) -> float:
        required_fields = schema.get('required', [])
        properties = schema.get('properties', {})
        
        required_present = sum(1 for field in required_fields if field in data and data[field] is not None)
        required_compliance = required_present / len(required_fields) if required_fields else 1.0
        
        type_errors = 0
        total_fields = 0
        for field, value in data.items():
            if field in properties:
                total_fields += 1
                expected_type = properties[field].get('type')
                if expected_type and not self._check_type(value, expected_type):
                    type_errors += 1
        
        type_compliance = 1.0 - (type_errors / total_fields) if total_fields > 0 else 1.0
        
        return (required_compliance * 0.7 + type_compliance * 0.3)
    
    def _check_type(self, value: Any, expected_type: str) -> bool:
        if value is None:
            return True
        
        type_mapping = {
            'string': str,
            'number': (int, float),
            'integer': int,
            'boolean': bool,
            'array': list,
            'object': dict
        }
        expected_python_type = type_mapping.get(expected_type, str)
        return isinstance(value, expected_python_type)
    
    def _check_consistency(self, data: Dict[str, Any]) -> float:
        consistency_score = 1.0
        
        if 'email' in data and data['email']:
            if '@' not in str(data['email']):
                consistency_score -= 0.1
        
        if 'startDate' in data and 'endDate' in data:
            try:
                if data['startDate'] and data['endDate']:
                    if str(data['startDate']) > str(data['endDate']):
                        consistency_score -= 0.15
            except:
                pass
        
        if isinstance(data, dict):
            for key, value in data.items():
                if isinstance(value, list):
                    for item in value:
                        if isinstance(item, dict):
                            consistency_score *= self._check_consistency(item)
                elif isinstance(value, dict):
                    consistency_score *= self._check_consistency(value)
        
        return max(0.7, consistency_score)
    
    def _generate_review_flags(self, field_scores: Dict[str, float], schema_compliance: float, overall_confidence: float, required_fields: List[str], extracted_data: Dict[str, Any], total_expected: int, extracted_count: int) -> List[str]:
        flags = []
        
        completeness_rate = extracted_count / total_expected if total_expected > 0 else 0
        
        if completeness_rate < 0.5:
            flags.append("incomplete_extraction")
        elif completeness_rate < 0.8:
            flags.append("partial_extraction")
        
        if overall_confidence < 0.6:
            flags.append("low_quality")
        elif overall_confidence < 0.8:
            flags.append("moderate_quality")
        
        if schema_compliance < 0.7:
            flags.append("schema_violations")
        
        missing_required = [field for field in required_fields if field not in extracted_data or extracted_data[field] is None]
        if missing_required:
            flags.append(f"missing_required_fields")
        
        empty_fields = [k for k, v in extracted_data.items() if v is None or v == ""]
        if len(empty_fields) > total_expected * 0.3:
            flags.append("many_empty_fields")
        
        return flags
    
    def _estimate_review_time(self, review_flags: List[str], field_scores: Dict[str, float]) -> int:
        if not review_flags:
            return 0
        
        low_confidence_count = len([score for score in field_scores.values() if score < 0.7])
        base_time = 5
        field_time = low_confidence_count * 2
        
        return min(base_time + field_time, 60)

class StructuredExtractionSystem:
    def __init__(self, api_key: str):
        self.schema_analyzer = SchemaAnalyzer()
        self.document_processor = DocumentProcessor()
        self.extraction_engine = ExtractionEngine(api_key)
        self.quality_assessor = QualityAssessor()
    
    async def extract_structured_data(
        self, 
        content: str, 
        schema: Dict[str, Any],
        options: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        start_time = datetime.now()
        
        logger.info("Starting structured data extraction")
        logger.info(f"Content length: {len(content)} characters")
        
        complexity = self.schema_analyzer.analyze_complexity(schema)
        logger.info(f"Schema complexity: Tier {complexity.complexity_tier}")
        
        plan = self.schema_analyzer.create_extraction_plan(schema, complexity)
        logger.info(f"Extraction plan: {len(plan.stages)} stages")
        
        chunks = self.document_processor.process_document(content, schema)
        logger.info(f"Document chunks: {len(chunks)}")
        
        result = await self.extraction_engine.extract(chunks[0], plan, schema)
        quality = self.quality_assessor.assess_extraction(result, schema)
        
        processing_time = (datetime.now() - start_time).total_seconds()
        
        logger.info(f"Extraction completed in {processing_time:.2f} seconds")
        logger.info(f"Overall confidence: {quality.overall_confidence:.3f}")
        
        return {
            "data": result.data,
            "confidence_scores": result.confidence_scores,
            "overall_confidence": quality.overall_confidence,
            "review_flags": quality.review_flags,
            "extraction_metadata": {
                "complexity_tier": complexity.complexity_tier,
                "stages_executed": len(plan.stages),
                "estimated_cost": plan.estimated_cost,
                "actual_processing_time": processing_time,
                "schema_compliance": quality.schema_compliance,
                "recommended_review_time": quality.recommended_review_time
            }
        }