Schema.org Extraction

# Extraction complète de Schema.org (JSON-LD, Microdata, RDFa)
import json
import re
from typing import List, Dict, Any
from selectolax.parser import HTMLParser

class SchemaExtractor:
    """Extracteur universel de données Schema.org"""
    
    def extract_jsonld(self, html: str) -> List[Dict[str, Any]]:
        """Extrait tous les blocs JSON-LD"""
        tree = HTMLParser(html)
        schemas = []
        
        for script in tree.css('script[type="application/ld+json"]'):
            try:
                text = script.text(strip=True)
                # Nettoyer les commentaires JS parfois présents
                text = re.sub(r'/*.*?*/', '', text, flags=re.DOTALL)
                data = json.loads(text)
                
                # Gérer les @graph (collections)
                if isinstance(data, dict) and '@graph' in data:
                    schemas.extend(data['@graph'])
                elif isinstance(data, list):
                    schemas.extend(data)
                else:
                    schemas.append(data)
            except json.JSONDecodeError:
                continue
        
        return schemas
    
    def find_by_type(self, schemas: List[Dict], schema_type: str) -> List[Dict]:
        """Filtre les schemas par @type"""
        results = []
        
        for schema in schemas:
            item_type = schema.get('@type', '')
            # @type peut être une string ou une liste
            if isinstance(item_type, list):
                if schema_type in item_type:
                    results.append(schema)
            elif item_type == schema_type:
                results.append(schema)
        
        return results
    
    def extract_products(self, html: str) -> List[Dict]:
        """Extrait et normalise les produits"""
        schemas = self.extract_jsonld(html)
        products = self.find_by_type(schemas, 'Product')
        
        normalized = []
        for product in products:
            # Extraire les offers (peut être imbriqué)
            offers = product.get('offers', {})
            if isinstance(offers, list):
                offers = offers[0] if offers else {}
            
            normalized.append({
                'name': product.get('name'),
                'description': product.get('description'),
                'sku': product.get('sku'),
                'brand': self._extract_brand(product),
                'price': offers.get('price'),
                'currency': offers.get('priceCurrency'),
                'availability': self._parse_availability(offers.get('availability', '')),
                'image': self._extract_image(product),
                'rating': self._extract_rating(product),
            })
        
        return normalized
    
    def _extract_brand(self, product: Dict) -> str:
        brand = product.get('brand', {})
        if isinstance(brand, dict):
            return brand.get('name', '')
        return brand or ''
    
    def _extract_image(self, product: Dict) -> str:
        image = product.get('image', '')
        if isinstance(image, list):
            return image[0] if image else ''
        if isinstance(image, dict):
            return image.get('url', '')
        return image
    
    def _extract_rating(self, product: Dict) -> Dict:
        rating = product.get('aggregateRating', {})
        return {
            'value': rating.get('ratingValue'),
            'count': rating.get('reviewCount'),
            'best': rating.get('bestRating', 5)
        }
    
    def _parse_availability(self, availability: str) -> bool:
        return 'InStock' in availability or 'PreOrder' in availability

# Usage
extractor = SchemaExtractor()
html = open('product_page.html').read()
products = extractor.extract_products(html)
print(json.dumps(products, indent=2))