┌─────────────────────────────────────────────────┐
│           AI Listing Optimization System         │
├─────────────────────────────────────────────────┤
│                                                  │
│  Data Collection → Processing → Analysis → Action│
│  (Pangolinfo API)  (ETL)       (AI/ML)   (Output)│
│                                                  │
└─────────────────────────────────────────────────┘

import requests
from typing import Dict, List, Optional
from dataclasses import dataclass
import time

@dataclass
class ProductData:
    asin: str
    title: str
    price: float
    rating: float
    review_count: int
    bullet_points: List[str]
    description: str
    rank: Optional[int] = None

class PangolinfoClient:
    """
    Wrapper for Pangolinfo Scrape API
    Handles rate limiting, retries, and error handling
    """

    def __init__(self, api_key: str, base_url: str = "https://api.pangolinfo.com"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })

    def scrape_product(self, asin: str, country: str = "us") -> ProductData:
        """Fetch detailed product data"""
        url = f"https://www.amazon.com/dp/{asin}"

        response = self._make_request({
            "url": url,
            "country": country,
            "output_format": "json"
        })

        return self._parse_product_data(response)

    def scrape_search_results(
        self, 
        keyword: str, 
        country: str = "us",
        page: int = 1
    ) -> List[ProductData]:
        """Fetch search result page"""
        url = f"https://www.amazon.com/s?k={keyword}&page={page}"

        response = self._make_request({
            "url": url,
            "country": country,
            "output_format": "json"
        })

        return [self._parse_product_data(p) for p in response.get('products', [])]

    def _make_request(self, payload: Dict) -> Dict:
        """Make API request with retry logic"""
        max_retries = 3

        for attempt in range(max_retries):
            try:
                response = self.session.post(
                    f"{self.base_url}/scrape",
                    json=payload,
                    timeout=30
                )
                response.raise_for_status()
                return response.json()

            except requests.exceptions.RequestException as e:
                if attempt == max_retries - 1:
                    raise

                wait_time = 2 ** attempt
                print(f"Request failed, retrying in {wait_time}s...")
                time.sleep(wait_time)

    def _parse_product_data(self, raw_data: Dict) -> ProductData:
        """Parse API response into ProductData object"""
        return ProductData(
            asin=raw_data.get('asin', ''),
            title=raw_data.get('title', ''),
            price=raw_data.get('price', {}).get('current', 0.0),
            rating=raw_data.get('rating', {}).get('average', 0.0),
            review_count=raw_data.get('rating', {}).get('count', 0),
            bullet_points=raw_data.get('bullet_points', []),
            description=raw_data.get('description', ''),
            rank=raw_data.get('rank', {}).get('position')
        )

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Callable

class BatchCollector:
    """Efficiently collect data for multiple products"""

    def __init__(self, client: PangolinfoClient, max_workers: int = 5):
        self.client = client
        self.max_workers = max_workers

    def collect_competitors(
        self, 
        keyword: str, 
        top_n: int = 20
    ) -> List[ProductData]:
        """Collect top N competitor data for a keyword"""

        # Step 1: Get search results
        search_results = []
        pages_needed = (top_n // 20) + 1

        for page in range(1, pages_needed + 1):
            results = self.client.scrape_search_results(keyword, page=page)
            search_results.extend(results)
            time.sleep(0.5)  # Rate limiting

        target_asins = [p.asin for p in search_results[:top_n]]

        # Step 2: Parallel fetch detailed data
        detailed_data = []

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {
                executor.submit(self.client.scrape_product, asin): asin
                for asin in target_asins
            }

            for future in as_completed(futures):
                try:
                    data = future.result()
                    detailed_data.append(data)
                except Exception as e:
                    print(f"Failed to collect {futures[future]}: {e}")

        return detailed_data

import re
from collections import Counter
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
import pandas as pd

class KeywordAnalyzer:
    """Extract and analyze keywords from competitor data"""

    def __init__(self):
        self.stop_words = set(stopwords.words('english'))
        # Add ecommerce-specific stop words
        self.stop_words.update(['amazon', 'brand', 'new', 'best'])

    def extract_keywords(self, text: str) -> List[str]:
        """Extract meaningful keywords from text"""
        # Lowercase and remove special chars
        text = re.sub(r'[^a-z0-9\s]', '', text.lower())

        # Tokenize
        tokens = word_tokenize(text)

        # Filter
        keywords = [
            word for word in tokens
            if word not in self.stop_words and len(word) > 2
        ]

        return keywords

    def analyze_competitors(
        self, 
        competitors: List[ProductData]
    ) -> pd.DataFrame:
        """Analyze keyword distribution across competitors"""

        all_keywords = []
        keyword_to_products = {}

        for product in competitors:
            # Extract from title and bullets
            text = f"{product.title} {' '.join(product.bullet_points)}"
            keywords = self.extract_keywords(text)

            all_keywords.extend(keywords)

            # Track which products use each keyword
            for kw in set(keywords):
                if kw not in keyword_to_products:
                    keyword_to_products[kw] = []
                keyword_to_products[kw].append(product.asin)

        # Calculate metrics
        keyword_freq = Counter(all_keywords)

        results = []
        for keyword, freq in keyword_freq.most_common(100):
            product_count = len(keyword_to_products[keyword])

            # Calculate average rank of products using this keyword
            avg_rank = sum(
                p.rank for p in competitors 
                if p.asin in keyword_to_products[keyword] and p.rank
            ) / product_count if product_count > 0 else 0

            results.append({
                'keyword': keyword,
                'frequency': freq,
                'product_count': product_count,
                'coverage': product_count / len(competitors),
                'avg_rank': avg_rank,
                'score': freq * (product_count / len(competitors))
            })

        return pd.DataFrame(results).sort_values('score', ascending=False)

class TitleOptimizer:
    """Analyze title patterns and generate recommendations"""

    def analyze_patterns(self, competitors: List[ProductData]) -> Dict:
        """Extract structural patterns from competitor titles"""

        patterns = {
            'avg_length': 0,
            'avg_word_count': 0,
            'common_structures': Counter(),
            'keyword_positions': {}
        }

        titles = [p.title for p in competitors]

        # Length analysis
        patterns['avg_length'] = sum(len(t) for t in titles) / len(titles)
        patterns['avg_word_count'] = sum(len(t.split()) for t in titles) / len(titles)

        # Structure analysis (simplified)
        for title in titles:
            words = title.split()
            if len(words) >= 5:
                structure = f"{words[0]}-{words[1]}-...-{words[-1]}"
                patterns['common_structures'][structure] += 1

        return patterns

    def generate_recommendations(
        self,
        keyword_analysis: pd.DataFrame,
        pattern_analysis: Dict,
        product_features: List[str]
    ) -> List[str]:
        """Generate optimized title candidates"""

        top_keywords = keyword_analysis.head(10)['keyword'].tolist()
        target_length = int(pattern_analysis['avg_length'])

        templates = [
            f"{product_features[0]} {top_keywords[0]} {top_keywords[1]} - {product_features[1]}",
            f"{top_keywords[0]} {top_keywords[1]} {product_features[0]} for {product_features[2]}",
            f"{product_features[0]} with {top_keywords[0]} and {top_keywords[1]} - {product_features[2]}"
        ]

        # Truncate to target length
        return [t[:target_length] for t in templates]

from celery import Celery
from celery.schedules import crontab

app = Celery('listing_optimizer', broker='redis://localhost:6379/0')

@app.task
def daily_competitor_analysis(keyword: str):
    """Daily task to analyze competitors"""

    client = PangolinfoClient(api_key=os.getenv('PANGOLINFO_API_KEY'))
    collector = BatchCollector(client)

    # Collect data
    competitors = collector.collect_competitors(keyword, top_n=20)

    # Analyze
    analyzer = KeywordAnalyzer()
    keyword_analysis = analyzer.analyze_competitors(competitors)

    # Store results
    store_analysis_results(keyword, keyword_analysis)

    # Check for significant changes
    if detect_market_shift(keyword_analysis):
        send_alert(f"Market shift detected for {keyword}")

@app.task
def monitor_specific_competitors(asins: List[str]):
    """Monitor specific competitor listings for changes"""

    client = PangolinfoClient(api_key=os.getenv('PANGOLINFO_API_KEY'))

    for asin in asins:
        current_data = client.scrape_product(asin)
        previous_data = load_previous_data(asin)

        if has_significant_change(current_data, previous_data):
            send_alert(f"Competitor {asin} updated listing")

        save_data(asin, current_data)

# Schedule tasks
app.conf.beat_schedule = {
    'daily-analysis': {
        'task': 'daily_competitor_analysis',
        'schedule': crontab(hour=2, minute=0),
        'args': ('bluetooth earbuds',)
    },
    'hourly-monitoring': {
        'task': 'monitor_specific_competitors',
        'schedule': crontab(minute=0),
        'args': (['B08XYZ1234', 'B09ABC5678'],)
    }
}

from time import time, sleep

class RateLimiter:
    def __init__(self, max_calls: int, period: int):
        self.max_calls = max_calls
        self.period = period
        self.calls = []

    def __call__(self, func):
        def wrapper(*args, **kwargs):
            now = time()
            self.calls = [c for c in self.calls if c > now - self.period]

            if len(self.calls) >= self.max_calls:
                sleep_time = self.period - (now - self.calls[0])
                sleep(sleep_time)

            self.calls.append(time())
            return func(*args, **kwargs)

        return wrapper

from functools import wraps
import logging

def retry_on_failure(max_retries=3, backoff_factor=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        logging.error(f"Failed after {max_retries} attempts: {e}")
                        raise

                    wait_time = backoff_factor ** attempt
                    logging.warning(f"Attempt {attempt + 1} failed, retrying in {wait_time}s")
                    sleep(wait_time)

        return wrapper
    return decorator

import redis
import json
from datetime import timedelta

class DataCache:
    def __init__(self, redis_client):
        self.redis = redis_client

    def get(self, key: str):
        data = self.redis.get(key)
        return json.loads(data) if data else None

    def set(self, key: str, value: dict, ttl: int = 3600):
        self.redis.setex(key, ttl, json.dumps(value))