Implementation Guide: Generate product descriptions, category pages, and email campaigns at scale

shopify auth login --store {store-name}.myshopify.com

wp wc product list --format=csv --fields=id,name,description,short_description,categories,attributes,images > product_catalog_export.csv --user=1

wp wc product list --format=csv --fields=id,name,slug,type,status,description,short_description,sku,price,regular_price,categories,tags,images,attributes --per_page=100 --user=1 > full_catalog.csv

# Strip HTML, flag thin content, and export products needing descriptions

# Save as clean_catalog.py
import pandas as pd
import re

df = pd.read_csv('full_catalog.csv')

# Remove HTML tags from existing descriptions
df['clean_description'] = df['description'].apply(lambda x: re.sub('<[^<]+?>', '', str(x)) if pd.notna(x) else '')

# Flag products with missing or thin descriptions (< 50 words)
df['word_count'] = df['clean_description'].apply(lambda x: len(str(x).split()))
df['needs_content'] = df['word_count'] < 50

# Summary
print(f'Total products: {len(df)}')
print(f'Products needing content: {df["needs_content"].sum()}')
print(f'Products with adequate content: {(~df["needs_content"]).sum()}')

# Export products needing content
df[df['needs_content']].to_csv('products_need_content.csv', index=False)
print(f'Exported {df["needs_content"].sum()} products to products_need_content.csv')

wp wc product update {product_id} --description='Generated description here' --user=1

pip install scikit-learn pandas

# detects near-duplicate generated descriptions using TF-IDF cosine
# similarity

import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

df = pd.read_csv('generated_descriptions.csv')

vectorizer = TfidfVectorizer(stop_words='english')
tfidf_matrix = vectorizer.fit_transform(df['generated_description'])

# Find pairs with >85% similarity
similarity_matrix = cosine_similarity(tfidf_matrix)
duplicates = []
for i in range(len(similarity_matrix)):
    for j in range(i+1, len(similarity_matrix)):
        if similarity_matrix[i][j] > 0.85:
            duplicates.append({
                'product_1': df.iloc[i]['product_name'],
                'product_2': df.iloc[j]['product_name'],
                'similarity': round(similarity_matrix[i][j], 3)
            })

dup_df = pd.DataFrame(duplicates)
print(f'Found {len(dup_df)} near-duplicate pairs')
dup_df.to_csv('duplicate_content_flags.csv', index=False)

#!/bin/bash
while IFS=, read -r product_id description; do
    wp wc product update $product_id --description="$description" --user=1
    echo "Updated product $product_id"
done < approved_descriptions.csv

wp db export backup_before_ai_content.sql

# system prompt, user prompt template, and OpenAI API integration

# Retail Product Description Generator - System Prompt

SYSTEM_PROMPT = """
You are an expert retail copywriter who creates compelling, accurate product descriptions for an online store.

## Brand Voice
{brand_voice_description}

## Rules
1. ACCURACY IS PARAMOUNT: Only describe features, materials, dimensions, and specifications that are explicitly provided in the product data. Never invent or assume specifications.
2. Include the primary SEO keyword naturally within the first 50 words.
3. Write in second person ('you') to speak directly to the customer.
4. Lead with the primary benefit or use case, not technical specs.
5. Include 3-5 key features as scannable bullet points.
6. End with a subtle call-to-action that encourages purchase.
7. Keep total length between {min_words} and {max_words} words.
8. Do NOT include pricing information unless explicitly instructed.
9. Do NOT make health, safety, or performance claims that aren't supported by the provided data.
10. Do NOT mention competitor products by name.
11. Generate a meta title (max 60 characters) and meta description (max 155 characters) optimized for the primary keyword.

## Output Format
Return a JSON object with these fields:
- "description_html": The product description in clean HTML (use <p>, <ul>, <li>, <strong> tags only)
- "short_description": A 1-2 sentence summary (max 50 words)
- "meta_title": SEO meta title (max 60 characters)
- "meta_description": SEO meta description (max 155 characters)
- "seo_keywords_used": List of keywords naturally incorporated
"""

USER_PROMPT_TEMPLATE = """
Generate a product description for the following item:

**Product Name:** {product_name}
**SKU:** {sku}
**Category:** {category}
**Primary SEO Keyword:** {primary_keyword}
**Secondary Keywords:** {secondary_keywords}
**Price:** {price}
**Key Attributes:**
{attributes}
**Additional Notes:** {notes}

Generate the product description following all rules in your instructions. Return valid JSON only.
"""

# Example usage with OpenAI API:
import openai
import json

def generate_product_description(
    product_data: dict,
    brand_voice: str,
    min_words: int = 150,
    max_words: int = 250,
    model: str = "gpt-4.1-mini"
) -> dict:
    """
    Generate a product description using OpenAI API.
    
    Args:
        product_data: Dict with keys: product_name, sku, category, primary_keyword,
                      secondary_keywords, price, attributes, notes
        brand_voice: String describing the brand's tone and style
        min_words: Minimum word count for description
        max_words: Maximum word count for description
        model: OpenAI model to use
    
    Returns:
        Dict with description_html, short_description, meta_title, meta_description
    """
    system = SYSTEM_PROMPT.format(
        brand_voice_description=brand_voice,
        min_words=min_words,
        max_words=max_words
    )
    
    # Format attributes as bullet points
    attributes_text = "\n".join(
        f"- {k}: {v}" for k, v in product_data.get('attributes', {}).items()
    )
    
    user_message = USER_PROMPT_TEMPLATE.format(
        product_name=product_data['product_name'],
        sku=product_data['sku'],
        category=product_data['category'],
        primary_keyword=product_data.get('primary_keyword', product_data['product_name']),
        secondary_keywords=', '.join(product_data.get('secondary_keywords', [])),
        price=product_data.get('price', 'Not specified'),
        attributes=attributes_text,
        notes=product_data.get('notes', 'None')
    )
    
    client = openai.OpenAI()  # Uses OPENAI_API_KEY env var
    
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user_message}
        ],
        response_format={"type": "json_object"},
        temperature=0.7,
        max_tokens=1000
    )
    
    result = json.loads(response.choices[0].message.content)
    return result

#!/usr/bin/env python3
"""
Bulk Product Content Generation Pipeline
Usage: python bulk_generate.py --input catalog.csv --output generated_content.csv --model gpt-4.1-mini
"""

import argparse
import csv
import json
import os
import time
import logging
from datetime import datetime
from pathlib import Path

import openai

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(f'bulk_generate_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Rate limiting configuration
RATE_LIMIT_RPM = 500  # Requests per minute for GPT-4.1-mini (Tier 1)
RATE_LIMIT_DELAY = 60.0 / RATE_LIMIT_RPM  # Seconds between requests
MAX_RETRIES = 3
RETRY_DELAY = 5  # Seconds

# Brand voice configuration (customize per client)
BRAND_VOICE = """
Our brand voice is warm, knowledgeable, and approachable. We speak like a trusted friend
who happens to be an expert. We use conversational language but maintain professionalism.
We focus on how products improve our customers' lives, not just features.
We avoid jargon, hyperbole, and aggressive sales language.
"""

SYSTEM_PROMPT = """You are an expert retail copywriter. Generate a product description following these rules:
1. Only describe features explicitly provided in the product data. Never invent specifications.
2. Include the primary SEO keyword naturally within the first 50 words.
3. Write in second person ('you').
4. Lead with the primary benefit, not technical specs.
5. Include 3-5 key features as bullet points.
6. End with a subtle call-to-action.
7. Keep between 150-250 words.
8. No health/safety claims unless supported by data.
9. No competitor mentions.

Brand voice: {brand_voice}

Return a JSON object with: description_html, short_description (max 50 words), meta_title (max 60 chars), meta_description (max 155 chars)"""


def generate_single_description(client: openai.OpenAI, product: dict, model: str) -> dict:
    """Generate a description for a single product with retry logic."""
    user_message = f"""Product: {product.get('product_name', 'Unknown')}
SKU: {product.get('sku', 'N/A')}
Category: {product.get('category', 'General')}
SEO Keyword: {product.get('primary_keyword', product.get('product_name', ''))}
Price: {product.get('price', 'Not specified')}
Attributes: {product.get('attributes', 'None provided')}
Image URL: {product.get('image_url', 'None')}

Generate the product description. Return valid JSON only."""

    for attempt in range(MAX_RETRIES):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": SYSTEM_PROMPT.format(brand_voice=BRAND_VOICE)},
                    {"role": "user", "content": user_message}
                ],
                response_format={"type": "json_object"},
                temperature=0.7,
                max_tokens=1000
            )
            
            result = json.loads(response.choices[0].message.content)
            result['tokens_used'] = response.usage.total_tokens
            result['model'] = model
            result['generated_at'] = datetime.now().isoformat()
            result['status'] = 'generated'
            return result
            
        except openai.RateLimitError:
            logger.warning(f"Rate limited on {product.get('sku')}. Waiting 60s...")
            time.sleep(60)
        except openai.APIError as e:
            logger.error(f"API error on {product.get('sku')}: {e}. Attempt {attempt+1}/{MAX_RETRIES}")
            time.sleep(RETRY_DELAY * (attempt + 1))
        except json.JSONDecodeError:
            logger.error(f"Invalid JSON response for {product.get('sku')}. Attempt {attempt+1}/{MAX_RETRIES}")
            time.sleep(RETRY_DELAY)
    
    return {'status': 'failed', 'error': f'Failed after {MAX_RETRIES} attempts', 'sku': product.get('sku')}


def process_catalog(input_file: str, output_file: str, model: str, batch_size: int = 100):
    """Process entire product catalog with progress tracking and checkpointing."""
    client = openai.OpenAI()  # Uses OPENAI_API_KEY env var
    
    # Read input catalog
    with open(input_file, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        products = list(reader)
    
    logger.info(f"Loaded {len(products)} products from {input_file}")
    
    # Check for existing checkpoint
    checkpoint_file = Path(output_file).with_suffix('.checkpoint.json')
    completed_skus = set()
    if checkpoint_file.exists():
        with open(checkpoint_file, 'r') as f:
            completed_skus = set(json.load(f))
        logger.info(f"Resuming from checkpoint. {len(completed_skus)} already completed.")
    
    # Process products
    results = []
    total_tokens = 0
    failed_count = 0
    
    for i, product in enumerate(products):
        sku = product.get('sku', f'row_{i}')
        
        # Skip if already completed
        if sku in completed_skus:
            continue
        
        logger.info(f"Processing {i+1}/{len(products)}: {product.get('product_name', 'Unknown')} ({sku})")
        
        result = generate_single_description(client, product, model)
        result['sku'] = sku
        result['product_name'] = product.get('product_name', '')
        results.append(result)
        
        if result.get('status') == 'generated':
            total_tokens += result.get('tokens_used', 0)
            completed_skus.add(sku)
        else:
            failed_count += 1
        
        # Rate limiting
        time.sleep(RATE_LIMIT_DELAY)
        
        # Checkpoint every batch_size products
        if (i + 1) % batch_size == 0:
            with open(checkpoint_file, 'w') as f:
                json.dump(list(completed_skus), f)
            logger.info(f"Checkpoint saved. {len(completed_skus)} completed, {failed_count} failed.")
            
            # Write intermediate results
            write_results(results, output_file)
    
    # Final write
    write_results(results, output_file)
    
    # Cost estimate
    est_cost = (total_tokens / 1_000_000) * 2.0  # Approximate blended rate for gpt-4.1-mini
    logger.info(f"\n=== COMPLETE ===")
    logger.info(f"Total products processed: {len(results)}")
    logger.info(f"Successful: {len(results) - failed_count}")
    logger.info(f"Failed: {failed_count}")
    logger.info(f"Total tokens used: {total_tokens:,}")
    logger.info(f"Estimated API cost: ${est_cost:.2f}")
    logger.info(f"Output saved to: {output_file}")


def write_results(results: list, output_file: str):
    """Write results to CSV for review."""
    fieldnames = ['sku', 'product_name', 'status', 'description_html', 'short_description',
                  'meta_title', 'meta_description', 'tokens_used', 'model', 'generated_at',
                  'reviewer', 'approved', 'notes']
    
    with open(output_file, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore')
        writer.writeheader()
        writer.writerows(results)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Bulk product description generator')
    parser.add_argument('--input', required=True, help='Input CSV file path')
    parser.add_argument('--output', default='generated_content.csv', help='Output CSV file path')
    parser.add_argument('--model', default='gpt-4.1-mini', help='OpenAI model to use')
    parser.add_argument('--batch-size', type=int, default=100, help='Checkpoint frequency')
    args = parser.parse_args()
    
    process_catalog(args.input, args.output, args.model, args.batch_size)

CATEGORY_PAGE_SYSTEM_PROMPT = """
You are an expert ecommerce content strategist and SEO copywriter. You create category landing pages
that serve dual purposes: (1) help shoppers find and choose the right products, and (2) rank well in
search engines for category-level keywords.

Brand voice: {brand_voice}

Rules:
1. Write between 300-500 words total.
2. Use the primary keyword in the H1, first paragraph, and at least one subheading.
3. Include secondary keywords naturally throughout.
4. Structure content with clear HTML headings (H1, H2) and paragraphs.
5. Include a 'How to Choose' or buying guide section with practical advice.
6. Mention 2-3 specific products from the category by name (provided in the data).
7. Write for the target audience specified.
8. Do NOT make unsubstantiated claims.
9. Include internal linking placeholders as [LINK: product-name] that can be replaced with actual URLs.

Return a JSON object with:
- "page_html": Full page content in clean HTML (H1, H2, p, ul, li, strong tags)
- "meta_title": SEO meta title (max 60 characters, include primary keyword)
- "meta_description": SEO meta description (max 155 characters, include primary keyword and CTA)
- "h1_tag": The recommended H1 heading text
- "word_count": Approximate word count of the page content
"""

CATEGORY_PAGE_USER_TEMPLATE = """
Generate a category landing page for:

**Category Name:** {category_name}
**Primary SEO Keyword:** {primary_keyword}
**Secondary Keywords:** {secondary_keywords}
**Number of Products:** {product_count}
**Price Range:** {price_range}
**Top Selling Products:** {top_products}
**Target Audience:** {target_audience}
**Season/Context:** {season_context}

Generate the full category page content. Return valid JSON only.
"""

# Usage: Call with model='gpt-4.1' for category pages (higher quality than mini)
# Temperature: 0.6 (slightly lower than product descriptions for more consistent structure)
# Max tokens: 2000

EMAIL_CAMPAIGN_SYSTEM_PROMPT = """
You are an expert email marketing copywriter for a retail brand. You create high-converting
email campaigns that drive opens, clicks, and purchases.

Brand voice: {brand_voice}

Rules:
1. Generate exactly 3 subject line variants for A/B testing (max 50 characters each).
2. Generate preview text (max 90 characters) that complements the subject line.
3. Email body should be concise — max 200 words. Retail emails should be scannable.
4. Include clear, action-oriented CTA text (e.g., 'Shop Now', 'See the Collection').
5. Use urgency or exclusivity sparingly and only when the campaign data warrants it.
6. Personalization tokens: use {{first_name}} for recipient name, {{product_name}} for dynamic product.
7. Do NOT make false urgency claims (e.g., don't say 'limited time' unless it actually is).
8. Include accessibility-friendly alt text suggestions for any image sections.
9. CAN-SPAM compliant: no deceptive subject lines.

Return a JSON object with:
- "subject_lines": Array of 3 subject line strings
- "preview_text": Preview/preheader text string
- "email_body_html": Email body in simple HTML (p, strong, a tags)
- "cta_text": Primary CTA button text
- "cta_url_placeholder": Suggested URL path (e.g., '/collections/summer-sale')
- "alt_texts": Array of suggested alt texts for image placeholders
"""

EMAIL_CAMPAIGN_USER_TEMPLATE = """
Generate email campaign content for:

**Campaign Type:** {campaign_type}  (e.g., 'New Product Launch', 'Seasonal Sale', 'Abandoned Cart', 'Post-Purchase', 'Weekly Newsletter')
**Campaign Name:** {campaign_name}
**Products Featured:** {featured_products}
**Offer/Promotion:** {offer_details}
**Target Segment:** {target_segment}
**Key Message:** {key_message}
**Urgency Level:** {urgency} (none / low / medium / high)
**Season/Context:** {season_context}

Generate the email campaign content. Return valid JSON only.
"""

# create a campaign using AI-generated content

import requests

def create_klaviyo_campaign_with_ai_content(
    klaviyo_api_key: str,
    list_id: str,
    ai_content: dict,
    from_email: str,
    from_name: str
):
    """
    Create a Klaviyo campaign using AI-generated content.
    
    Prerequisites:
    - Klaviyo API key with campaign:write scope
    - Template already created in Klaviyo with content blocks
    """
    headers = {
        'Authorization': f'Klaviyo-API-Key {klaviyo_api_key}',
        'Content-Type': 'application/json',
        'revision': '2024-10-15'
    }
    
    # Create campaign
    campaign_payload = {
        'data': {
            'type': 'campaign',
            'attributes': {
                'name': f'AI Generated - {ai_content.get("campaign_name", "Campaign")}',
                'audiences': {
                    'included': [{'type': 'list', 'id': list_id}]
                },
                'send_strategy': {
                    'method': 'static',
                    'options_static': {'send_time': None}  # Set send time as needed
                },
                'channel': 'email'
            }
        }
    }
    
    response = requests.post(
        'https://a.klaviyo.com/api/campaigns/',
        headers=headers,
        json=campaign_payload
    )
    
    if response.status_code == 201:
        campaign_id = response.json()['data']['id']
        print(f'Campaign created: {campaign_id}')
        print(f'Subject line options: {ai_content["subject_lines"]}')
        print(f'Preview text: {ai_content["preview_text"]}')
        return campaign_id
    else:
        print(f'Error: {response.status_code} - {response.text}')
        return None

#!/usr/bin/env python3
"""
Shopify Content Sync Integration
Pushes approved AI-generated content to Shopify via Admin API.

Requirements:
    pip install shopify requests python-dotenv

Environment variables (.env):
    SHOPIFY_STORE_URL=your-store.myshopify.com
    SHOPIFY_ACCESS_TOKEN=shpat_xxxxxxxxxxxxx

Usage:
    python shopify_sync.py --input approved_content.csv --dry-run
    python shopify_sync.py --input approved_content.csv --execute
"""

import argparse
import csv
import json
import os
import time
import logging
from datetime import datetime
from pathlib import Path

import requests
from dotenv import load_dotenv

load_dotenv()
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Shopify API rate limit: 2 requests/second for standard plans
SHOPIFY_RATE_LIMIT_DELAY = 0.55  # seconds between requests


class ShopifyContentSync:
    def __init__(self, store_url: str, access_token: str):
        self.base_url = f'https://{store_url}/admin/api/2024-10'
        self.headers = {
            'X-Shopify-Access-Token': access_token,
            'Content-Type': 'application/json'
        }
        self.backup = []  # Store original content for rollback
    
    def get_product_by_sku(self, sku: str) -> dict | None:
        """Find a Shopify product by SKU via variant lookup."""
        resp = requests.get(
            f'{self.base_url}/variants.json?sku={sku}&fields=id,product_id,sku',
            headers=self.headers
        )
        time.sleep(SHOPIFY_RATE_LIMIT_DELAY)
        
        if resp.status_code == 200 and resp.json().get('variants'):
            product_id = resp.json()['variants'][0]['product_id']
            product_resp = requests.get(
                f'{self.base_url}/products/{product_id}.json',
                headers=self.headers
            )
            time.sleep(SHOPIFY_RATE_LIMIT_DELAY)
            if product_resp.status_code == 200:
                return product_resp.json().get('product')
        return None
    
    def update_product_content(self, product_id: int, content: dict, dry_run: bool = True) -> bool:
        """Update product description, meta title, and meta description."""
        payload = {
            'product': {
                'id': product_id,
                'body_html': content.get('description_html', ''),
                'metafields_global_title_tag': content.get('meta_title', ''),
                'metafields_global_description_tag': content.get('meta_description', '')
            }
        }
        
        if dry_run:
            logger.info(f'  [DRY RUN] Would update product {product_id}')
            return True
        
        resp = requests.put(
            f'{self.base_url}/products/{product_id}.json',
            headers=self.headers,
            json=payload
        )
        time.sleep(SHOPIFY_RATE_LIMIT_DELAY)
        
        if resp.status_code == 200:
            logger.info(f'  Updated product {product_id} successfully')
            return True
        else:
            logger.error(f'  Failed to update product {product_id}: {resp.status_code} {resp.text}')
            return False
    
    def backup_product(self, product: dict):
        """Save original product content for rollback."""
        self.backup.append({
            'product_id': product['id'],
            'body_html': product.get('body_html', ''),
            'title': product.get('title', ''),
            'backed_up_at': datetime.now().isoformat()
        })
    
    def save_backup(self, filepath: str):
        """Save backup to JSON file."""
        with open(filepath, 'w') as f:
            json.dump(self.backup, f, indent=2)
        logger.info(f'Backup saved to {filepath} ({len(self.backup)} products)')
    
    def rollback(self, backup_filepath: str):
        """Restore original content from backup."""
        with open(backup_filepath, 'r') as f:
            backup_data = json.load(f)
        
        for item in backup_data:
            payload = {'product': {'id': item['product_id'], 'body_html': item['body_html']}}
            resp = requests.put(
                f'{self.base_url}/products/{item["product_id"]}.json',
                headers=self.headers,
                json=payload
            )
            time.sleep(SHOPIFY_RATE_LIMIT_DELAY)
            if resp.status_code == 200:
                logger.info(f'Rolled back product {item["product_id"]}')
            else:
                logger.error(f'Failed to rollback product {item["product_id"]}')


def sync_approved_content(input_file: str, dry_run: bool = True):
    """Main sync function: reads approved CSV and pushes to Shopify."""
    store_url = os.getenv('SHOPIFY_STORE_URL')
    access_token = os.getenv('SHOPIFY_ACCESS_TOKEN')
    
    if not store_url or not access_token:
        raise ValueError('Set SHOPIFY_STORE_URL and SHOPIFY_ACCESS_TOKEN in .env')
    
    sync = ShopifyContentSync(store_url, access_token)
    
    with open(input_file, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        rows = [r for r in reader if r.get('approved', '').lower() == 'yes']
    
    logger.info(f'Found {len(rows)} approved products to sync')
    
    success = 0
    failed = 0
    
    for row in rows:
        sku = row.get('sku', '')
        logger.info(f'Processing SKU: {sku} - {row.get("product_name", "")}')
        
        product = sync.get_product_by_sku(sku)
        if not product:
            logger.warning(f'  Product not found for SKU: {sku}')
            failed += 1
            continue
        
        sync.backup_product(product)
        
        if sync.update_product_content(product['id'], row, dry_run=dry_run):
            success += 1
        else:
            failed += 1
    
    # Save backup
    backup_path = f'backup_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json'
    sync.save_backup(backup_path)
    
    logger.info(f'\nSync complete: {success} updated, {failed} failed')
    logger.info(f'Backup saved to: {backup_path}')
    if dry_run:
        logger.info('This was a DRY RUN. Re-run with --execute to apply changes.')


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Sync AI content to Shopify')
    parser.add_argument('--input', required=True, help='Approved content CSV')
    parser.add_argument('--dry-run', action='store_true', default=True, help='Preview without changes')
    parser.add_argument('--execute', action='store_true', help='Apply changes to live store')
    parser.add_argument('--rollback', help='Rollback using backup JSON file')
    args = parser.parse_args()
    
    if args.rollback:
        sync = ShopifyContentSync(os.getenv('SHOPIFY_STORE_URL'), os.getenv('SHOPIFY_ACCESS_TOKEN'))
        sync.rollback(args.rollback)
    else:
        sync_approved_content(args.input, dry_run=not args.execute)

#!/usr/bin/env python3
"""
Content Quality Validation Agent
Validates AI-generated content against source product data.

Runs as a post-generation QA step, flagging issues before human review.

Usage: python qa_agent.py --products catalog.csv --content generated_content.csv --output qa_report.csv
"""

import csv
import json
import re
import logging
from dataclasses import dataclass, field

import openai

logger = logging.getLogger(__name__)

@dataclass
class QAResult:
    sku: str
    product_name: str
    issues: list = field(default_factory=list)
    severity: str = 'pass'  # pass, warning, fail
    score: float = 1.0


def check_word_count(description: str, min_words: int = 100, max_words: int = 300) -> list:
    """Check if description meets word count requirements."""
    word_count = len(description.split())
    issues = []
    if word_count < min_words:
        issues.append(f'Too short: {word_count} words (minimum {min_words})')
    if word_count > max_words:
        issues.append(f'Too long: {word_count} words (maximum {max_words})')
    return issues


def check_meta_lengths(meta_title: str, meta_description: str) -> list:
    """Check SEO meta field lengths."""
    issues = []
    if len(meta_title) > 60:
        issues.append(f'Meta title too long: {len(meta_title)} chars (max 60)')
    if len(meta_title) < 20:
        issues.append(f'Meta title too short: {len(meta_title)} chars (min 20)')
    if len(meta_description) > 155:
        issues.append(f'Meta description too long: {len(meta_description)} chars (max 155)')
    if len(meta_description) < 70:
        issues.append(f'Meta description too short: {len(meta_description)} chars (min 70)')
    return issues


def check_prohibited_patterns(text: str) -> list:
    """Check for prohibited content patterns."""
    issues = []
    # Check for fake urgency without campaign context
    urgency_phrases = ['limited time only', 'act now', 'don\'t miss out', 'while supplies last',
                       'hurry', 'only \\d+ left', 'selling fast']
    for phrase in urgency_phrases:
        if re.search(phrase, text.lower()):
            issues.append(f'Contains urgency language: "{phrase}" - verify this is appropriate')
    
    # Check for superlatives that may be unsubstantiated
    superlatives = ['best', 'finest', 'greatest', '#1', 'number one', 'world\'s', 'industry-leading']
    for word in superlatives:
        if word in text.lower():
            issues.append(f'Contains superlative "{word}" - verify this claim is substantiated')
    
    # Check for health/safety claims
    health_words = ['cure', 'treat', 'heal', 'prevent disease', 'clinically proven',
                    'doctor recommended', 'FDA approved', 'medical grade']
    for phrase in health_words:
        if phrase in text.lower():
            issues.append(f'CRITICAL: Contains health/safety claim "{phrase}" - requires verification')
    
    return issues


def check_specification_accuracy(description: str, product_data: dict) -> list:
    """Use AI to verify that description doesn't hallucinate specs not in source data."""
    client = openai.OpenAI()
    
    verification_prompt = f"""Compare this product description against the source product data.
Identify any claims, specifications, or features in the description that are NOT present in the source data.

Source Product Data:
{json.dumps(product_data, indent=2)}

Generated Description:
{description}

Return a JSON object with:
- "hallucinated_claims": Array of specific claims in the description not supported by source data
- "accuracy_score": Float 0-1 (1 = perfectly accurate, 0 = completely fabricated)
- "notes": Any additional observations
"""
    
    try:
        response = client.chat.completions.create(
            model='gpt-4.1-mini',
            messages=[{'role': 'user', 'content': verification_prompt}],
            response_format={'type': 'json_object'},
            temperature=0.1,
            max_tokens=500
        )
        result = json.loads(response.choices[0].message.content)
        
        issues = []
        for claim in result.get('hallucinated_claims', []):
            issues.append(f'Potential hallucination: {claim}')
        
        return issues, result.get('accuracy_score', 0.5)
    except Exception as e:
        logger.error(f'Specification check failed: {e}')
        return [f'Specification check error: {str(e)}'], 0.5


def run_qa_pipeline(products_file: str, content_file: str, output_file: str):
    """Run full QA pipeline on generated content."""
    # Load source product data
    with open(products_file, 'r', encoding='utf-8') as f:
        products = {row['sku']: row for row in csv.DictReader(f)}
    
    # Load generated content
    with open(content_file, 'r', encoding='utf-8') as f:
        content_rows = list(csv.DictReader(f))
    
    logger.info(f'Running QA on {len(content_rows)} generated descriptions')
    
    results = []
    for row in content_rows:
        sku = row.get('sku', '')
        qa = QAResult(sku=sku, product_name=row.get('product_name', ''))
        
        description = row.get('description_html', '')
        plain_text = re.sub('<[^<]+?>', '', description)  # Strip HTML
        
        # Run checks
        qa.issues.extend(check_word_count(plain_text))
        qa.issues.extend(check_meta_lengths(
            row.get('meta_title', ''),
            row.get('meta_description', '')
        ))
        qa.issues.extend(check_prohibited_patterns(plain_text))
        
        # Specification accuracy check (uses AI)
        if sku in products:
            spec_issues, accuracy = check_specification_accuracy(plain_text, products[sku])
            qa.issues.extend(spec_issues)
            qa.score = accuracy
        
        # Determine severity
        critical_issues = [i for i in qa.issues if 'CRITICAL' in i or 'hallucination' in i.lower()]
        if critical_issues:
            qa.severity = 'fail'
        elif qa.issues:
            qa.severity = 'warning'
        else:
            qa.severity = 'pass'
        
        results.append(qa)
    
    # Write QA report
    with open(output_file, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['SKU', 'Product Name', 'Severity', 'Accuracy Score', 'Issues'])
        for r in results:
            writer.writerow([r.sku, r.product_name, r.severity, f'{r.score:.2f}', ' | '.join(r.issues)])
    
    # Summary
    pass_count = sum(1 for r in results if r.severity == 'pass')
    warn_count = sum(1 for r in results if r.severity == 'warning')
    fail_count = sum(1 for r in results if r.severity == 'fail')
    
    logger.info(f'\nQA Report Summary:')
    logger.info(f'  PASS: {pass_count}')
    logger.info(f'  WARNING: {warn_count}')
    logger.info(f'  FAIL: {fail_count}')
    logger.info(f'Report saved to: {output_file}')


if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument('--products', required=True)
    parser.add_argument('--content', required=True)
    parser.add_argument('--output', default='qa_report.csv')
    args = parser.parse_args()
    run_qa_pipeline(args.products, args.content, args.output)

# run against all generated descriptions in the same category

python check_duplicates.py

# process 100 products; record total time, API errors, cost, and success
# rate

python bulk_generate.py

# verify 10 approved products are matched by SKU and report what would be
# updated

python shopify_sync.py --dry-run

# execute against 3 test products and verify descriptions, HTML rendering,
# and meta tags on the live store

python shopify_sync.py --execute

# run against test set of 3 accurate and 2 hallucinated descriptions

python qa_agent.py

# revert the 3 live test products to their original descriptions using the
# backup file

python shopify_sync.py --rollback backup_YYYYMMDD.json