Automate Product Descriptions 🚀

# Production Pipeline with Shopify Integration (500-5000 products)
from langgraph.graph import Graph, END
from typing import TypedDict, List, Dict
import openai
import shopify
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
import logging

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

class ProductState(TypedDict):
    product_id: str
    raw_data: Dict
    extracted_attributes: Dict
    descriptions: Dict
    seo_optimization: Dict
    shopify_formatted: Dict
    status: str
    retry_count: int
    errors: List[str]

class DescriptionPipeline:
    def __init__(self, shopify_shop_url: str, shopify_token: str, openai_key: str):
        self.shopify_session = shopify.Session(shopify_shop_url, '2024-01', shopify_token)
        shopify.ShopifyResource.activate_session(self.shopify_session)
        openai.api_key = openai_key
        self.executor = ThreadPoolExecutor(max_workers=20)
        self.graph = self.build_pipeline()
    
    def extract_node(self, state: ProductState) -> ProductState:
        """Extract and structure product attributes"""
        try:
            response = openai.chat.completions.create(
                model="gpt-4",
                messages=[{
                    "role": "user",
                    "content": f"Extract product attributes as JSON: {state['raw_data']}"
                }],
                temperature=0.3
            )
            
            state['extracted_attributes'] = json.loads(response.choices[0].message.content)
            state['status'] = 'extracted'
            logger.info(f"Extracted: {state['product_id']}")
        except Exception as e:
            state['errors'].append(f"Extraction failed: {str(e)}")
            state['retry_count'] += 1
        
        return state
    
    def generate_node(self, state: ProductState) -> ProductState:
        """Generate descriptions with category-specific templates"""
        try:
            category = state['extracted_attributes'].get('category', 'general')
            template = self.get_category_template(category)
            
            response = openai.chat.completions.create(
                model="gpt-4",
                messages=[
                    {"role": "system", "content": template},
                    {
                        "role": "user",
                        "content": f"Generate complete descriptions: {json.dumps(state['extracted_attributes'])}"
                    }
                ],
                temperature=0.7
            )
            
            state['descriptions'] = json.loads(response.choices[0].message.content)
            state['status'] = 'generated'
            logger.info(f"Generated: {state['product_id']}")
        except Exception as e:
            state['errors'].append(f"Generation failed: {str(e)}")
            state['retry_count'] += 1
        
        return state
    
    def optimize_node(self, state: ProductState) -> ProductState:
        """SEO optimization and keyword analysis"""
        try:
            response = openai.chat.completions.create(
                model="gpt-4",
                messages=[{
                    "role": "user",
                    "content": f"""Optimize for SEO:
Description: {state['descriptions']['long_description']}
Keywords: {state['descriptions']['seo_keywords']}

Provide: primary_keyword, secondary_keywords, keyword_density, readability_score, cta"""
                }],
                temperature=0.3
            )
            
            state['seo_optimization'] = json.loads(response.choices[0].message.content)
            state['status'] = 'optimized'
            logger.info(f"Optimized: {state['product_id']}")
        except Exception as e:
            state['errors'].append(f"Optimization failed: {str(e)}")
            state['retry_count'] += 1
        
        return state
    
    def format_shopify_node(self, state: ProductState) -> ProductState:
        """Format for Shopify API"""
        try:
            state['shopify_formatted'] = {
                'product': {
                    'id': state['product_id'],
                    'body_html': state['descriptions']['long_description'],
                    'title': state['extracted_attributes']['product_name'],
                    'product_type': state['extracted_attributes']['category'],
                    'tags': ', '.join(state['descriptions']['seo_keywords']),
                    'metafields': [
                        {
                            'namespace': 'seo',
                            'key': 'meta_title',
                            'value': state['descriptions']['meta_title'],
                            'type': 'single_line_text_field'
                        },
                        {
                            'namespace': 'seo',
                            'key': 'meta_description',
                            'value': state['descriptions']['meta_description'],
                            'type': 'single_line_text_field'
                        }
                    ]
                }
            }
            state['status'] = 'formatted'
            logger.info(f"Formatted: {state['product_id']}")
        except Exception as e:
            state['errors'].append(f"Formatting failed: {str(e)}")
            state['retry_count'] += 1
        
        return state
    
    def upload_shopify_node(self, state: ProductState) -> ProductState:
        """Upload to Shopify"""
        try:
            product = shopify.Product.find(state['product_id'])
            product.body_html = state['shopify_formatted']['product']['body_html']
            product.tags = state['shopify_formatted']['product']['tags']
            
            # Update metafields
            for metafield_data in state['shopify_formatted']['product']['metafields']:
                metafield = shopify.Metafield(metafield_data)
                metafield.owner_id = state['product_id']
                metafield.owner_resource = 'product'
                metafield.save()
            
            product.save()
            state['status'] = 'uploaded'
            logger.info(f"Uploaded: {state['product_id']}")
        except Exception as e:
            state['errors'].append(f"Upload failed: {str(e)}")
            state['retry_count'] += 1
        
        return state
    
    def check_status(self, state: ProductState) -> str:
        """Route based on status and retries"""
        if state['status'] == 'uploaded':
            return "complete"
        elif state['retry_count'] >= 3:
            return "failed"
        elif len(state['errors']) > 0:
            return "retry"
        else:
            return "continue"
    
    def get_category_template(self, category: str) -> str:
        """Get category-specific writing template"""
        templates = {
            'yoga': "You're writing for wellness-conscious consumers. Tone: Calm, inspiring, eco-focused.",
            'electronics': "You're writing for tech buyers. Tone: Professional, detailed, spec-focused.",
            'fashion': "You're writing for style shoppers. Tone: Trendy, aspirational, style-focused.",
            'home': "You're writing for home decorators. Tone: Cozy, practical, design-focused."
        }
        return templates.get(category, "You're writing product descriptions. Tone: Professional, engaging.")
    
    def build_pipeline(self):
        """Build the LangGraph pipeline"""
        graph = Graph()
        
        # Add nodes
        graph.add_node("extract", self.extract_node)
        graph.add_node("generate", self.generate_node)
        graph.add_node("optimize", self.optimize_node)
        graph.add_node("format_shopify", self.format_shopify_node)
        graph.add_node("upload_shopify", self.upload_shopify_node)
        
        # Add edges
        graph.set_entry_point("extract")
        graph.add_edge("extract", "generate")
        graph.add_edge("generate", "optimize")
        graph.add_edge("optimize", "format_shopify")
        graph.add_edge("format_shopify", "upload_shopify")
        
        # Conditional routing
        graph.add_conditional_edges(
            "upload_shopify",
            self.check_status,
            {
                "complete": END,
                "failed": END,
                "retry": "extract",
                "continue": "upload_shopify"
            }
        )
        
        return graph.compile()
    
    async def process_batch(self, product_ids: List[str]) -> List[ProductState]:
        """Process multiple products in parallel"""
        results = []
        
        # Create initial states
        tasks = []
        for product_id in product_ids:
            # Fetch product from Shopify
            product = shopify.Product.find(product_id)
            
            initial_state = {
                'product_id': product_id,
                'raw_data': product.to_dict(),
                'extracted_attributes': {},
                'descriptions': {},
                'seo_optimization': {},
                'shopify_formatted': {},
                'status': 'pending',
                'retry_count': 0,
                'errors': []
            }
            
            # Process through pipeline
            task = asyncio.create_task(self.process_single(initial_state))
            tasks.append(task)
        
        # Wait for all to complete
        results = await asyncio.gather(*tasks)
        
        # Log summary
        successful = len([r for r in results if r['status'] == 'uploaded'])
        failed = len([r for r in results if r['status'] == 'failed'])
        logger.info(f"Batch complete: {successful} successful, {failed} failed")
        
        return results
    
    async def process_single(self, state: ProductState) -> ProductState:
        """Process single product through pipeline"""
        return self.graph.invoke(state)

# Usage
pipeline = DescriptionPipeline(
    shopify_shop_url="your-store.myshopify.com",
    shopify_token="shpat_your_token",
    openai_key="sk-proj-your-key"
)

# Get all products without descriptions
products = shopify.Product.find(fields='id', body_html=None)
product_ids = [p.id for p in products]

# Process in batches of 50
batch_size = 50
for i in range(0, len(product_ids), batch_size):
    batch = product_ids[i:i+batch_size]
    results = asyncio.run(pipeline.process_batch(batch))
    print(f"Processed batch {i//batch_size + 1}: {len(results)} products")

print(f"Complete! Processed {len(product_ids)} products")