Netflix Streaming.

# streaming/abr_client.py

"""
Adaptive Bitrate Streaming Client
Demonstrates how Netflix clients select quality levels dynamically.
"""

from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
import time


class QualityLevel(Enum):
    """Available quality levels."""
    Q_240P = (240, 235_000)    # (height, bitrate_bps)
    Q_360P = (360, 560_000)
    Q_480P = (480, 1_000_000)
    Q_720P = (720, 3_000_000)
    Q_1080P = (1080, 5_000_000)
    Q_4K = (2160, 15_000_000)


@dataclass
class ChunkInfo:
    """Information about a video chunk."""
    index: int
    duration_seconds: float
    quality: QualityLevel
    size_bytes: int
    url: str


@dataclass
class BufferState:
    """Current state of playback buffer."""
    buffered_seconds: float
    current_position: float
    is_playing: bool


class ABRController:
    """
    Adaptive Bitrate Controller.
    
    Implements a buffer-based ABR algorithm similar to Netflix's approach.
    Balances between:
    - Maximizing video quality
    - Avoiding rebuffering
    - Minimizing quality switches
    
    Applies concepts from:
    - Week 2: Timeout management and graceful degradation
    - Week 4: Caching and buffer management
    """
    
    # Buffer thresholds (seconds)
    BUFFER_MIN = 5.0      # Minimum buffer before panic
    BUFFER_LOW = 15.0     # Start being conservative
    BUFFER_TARGET = 30.0  # Ideal buffer level
    BUFFER_MAX = 60.0     # Maximum buffer
    
    # Quality switch dampening
    SWITCH_COOLDOWN = 10.0  # Seconds between quality changes
    
    def __init__(self, quality_levels: List[QualityLevel]):
        self.quality_levels = sorted(
            quality_levels, 
            key=lambda q: q.value[1]  # Sort by bitrate
        )
        self.current_quality_idx = 0
        self.last_switch_time = 0.0
        self.bandwidth_estimates: List[float] = []
        
    def estimate_bandwidth(self, chunk_size: int, download_time: float) -> float:
        """
        Estimate available bandwidth from chunk download.
        
        Uses a harmonic mean of recent samples to smooth estimates.
        """
        if download_time <= 0:
            return 0
            
        measured_bps = (chunk_size * 8) / download_time
        
        # Keep last 5 measurements
        self.bandwidth_estimates.append(measured_bps)
        if len(self.bandwidth_estimates) > 5:
            self.bandwidth_estimates.pop(0)
        
        # Use harmonic mean (conservative estimate)
        if not self.bandwidth_estimates:
            return measured_bps
            
        harmonic_mean = len(self.bandwidth_estimates) / sum(
            1/bw for bw in self.bandwidth_estimates
        )
        return harmonic_mean
    
    def select_quality(
        self, 
        buffer_state: BufferState,
        estimated_bandwidth: float,
        current_time: float
    ) -> QualityLevel:
        """
        Select optimal quality level based on buffer and bandwidth.
        
        Algorithm:
        1. If buffer critically low → drop to lowest quality
        2. If buffer low → be conservative, don't increase
        3. If buffer healthy → select highest sustainable quality
        4. Apply switch dampening to avoid oscillation
        """
        buffer = buffer_state.buffered_seconds
        
        # PANIC MODE: Buffer critically low
        if buffer < self.BUFFER_MIN:
            self.current_quality_idx = 0
            self.last_switch_time = current_time
            return self.quality_levels[0]
        
        # Find highest sustainable quality
        target_idx = 0
        for i, quality in enumerate(self.quality_levels):
            bitrate = quality.value[1]
            # Need 20% headroom for safety
            if bitrate * 1.2 <= estimated_bandwidth:
                target_idx = i
        
        # Buffer-based adjustment
        if buffer < self.BUFFER_LOW:
            # Don't increase quality when buffer is low
            target_idx = min(target_idx, self.current_quality_idx)
        elif buffer > self.BUFFER_TARGET:
            # Allow quality increase when buffer is healthy
            pass
        
        # Switch dampening: avoid rapid oscillation
        time_since_switch = current_time - self.last_switch_time
        if time_since_switch < self.SWITCH_COOLDOWN:
            # Only allow quality decrease, not increase
            target_idx = min(target_idx, self.current_quality_idx)
        
        # Apply change
        if target_idx != self.current_quality_idx:
            self.current_quality_idx = target_idx
            self.last_switch_time = current_time
            
        return self.quality_levels[self.current_quality_idx]
    
    def get_initial_quality(self) -> QualityLevel:
        """
        Select initial quality for playback start.
        
        Strategy: Start at medium quality for fast startup,
        then adapt based on actual bandwidth.
        """
        # Start at 480p - reasonable quality, fast start
        for i, q in enumerate(self.quality_levels):
            if q.value[0] >= 480:
                self.current_quality_idx = i
                return q
        return self.quality_levels[0]


class StreamingSession:
    """
    Manages a complete streaming session.
    
    Coordinates:
    - Chunk downloading
    - Buffer management
    - Quality selection
    - Playback control
    """
    
    def __init__(self, manifest_url: str):
        self.manifest_url = manifest_url
        self.abr = ABRController(list(QualityLevel))
        self.buffer = BufferState(
            buffered_seconds=0.0,
            current_position=0.0,
            is_playing=False
        )
        self.chunks_downloaded = 0
        self.total_rebuffers = 0
        self.quality_switches = 0
        
    async def start_playback(self):
        """
        Initialize playback session.
        
        1. Fetch manifest
        2. Select initial quality
        3. Pre-buffer minimum amount
        4. Start playback
        """
        # Start with medium quality for fast startup
        initial_quality = self.abr.get_initial_quality()
        
        # Pre-buffer 5 seconds before starting
        while self.buffer.buffered_seconds < 5.0:
            await self._download_next_chunk(initial_quality)
        
        self.buffer.is_playing = True
        
    async def _download_next_chunk(self, quality: QualityLevel):
        """Download next chunk at specified quality."""
        # Simulated download - in reality would fetch from CDN
        chunk_size = quality.value[1] * 4 // 8  # 4-second chunk, in bytes
        download_time = (chunk_size * 8) / 5_000_000  # Simulate 5 Mbps
        
        await self._simulate_download(download_time)
        
        self.buffer.buffered_seconds += 4.0
        self.chunks_downloaded += 1
        
        # Update bandwidth estimate
        self.abr.estimate_bandwidth(chunk_size, download_time)
        
    async def _simulate_download(self, seconds: float):
        """Simulate network download time."""
        import asyncio
        await asyncio.sleep(seconds)
    
    def get_session_stats(self) -> dict:
        """Return session quality metrics."""
        return {
            "chunks_downloaded": self.chunks_downloaded,
            "rebuffer_events": self.total_rebuffers,
            "quality_switches": self.quality_switches,
            "current_quality": self.abr.quality_levels[
                self.abr.current_quality_idx
            ].name
        }

# cdn/content_placement.py

"""
Content Placement Algorithm for Open Connect
Determines which content to cache on which OCA servers.
"""

from dataclasses import dataclass
from typing import Dict, List, Optional, Set
from datetime import datetime, timedelta
import heapq


@dataclass
class ContentItem:
    """A piece of content in the catalog."""
    content_id: str
    size_bytes: int
    popularity_score: float
    release_date: datetime
    regions: List[str]  # Regions where content is available


@dataclass
class OCAServer:
    """An Open Connect Appliance server."""
    server_id: str
    location: str
    region: str
    capacity_bytes: int
    used_bytes: int
    cached_content: Set[str]
    
    @property
    def available_bytes(self) -> int:
        return self.capacity_bytes - self.used_bytes


class ContentPlacementEngine:
    """
    Determines optimal content placement across OCA fleet.
    
    Goals:
    1. Maximize cache hit rate
    2. Minimize origin fetches
    3. Balance load across servers
    4. Account for regional popularity differences
    
    Applies concepts from:
    - Week 1: Partitioning and replication strategies
    - Week 4: Cache placement and invalidation
    """
    
    def __init__(self):
        self.servers: Dict[str, OCAServer] = {}
        self.content: Dict[str, ContentItem] = {}
        self.regional_popularity: Dict[str, Dict[str, float]] = {}
        
    def add_server(self, server: OCAServer):
        """Register an OCA server."""
        self.servers[server.server_id] = server
        
    def update_popularity(self, region: str, content_id: str, score: float):
        """Update regional popularity score for content."""
        if region not in self.regional_popularity:
            self.regional_popularity[region] = {}
        self.regional_popularity[region][content_id] = score
    
    def compute_placement(self, region: str) -> Dict[str, List[str]]:
        """
        Compute optimal content placement for a region.
        
        Algorithm:
        1. Get all servers in region
        2. Rank content by regional popularity
        3. Place highest-popularity content on all servers
        4. Fill remaining space with long-tail content
        
        Returns: Dict mapping server_id to list of content_ids
        """
        regional_servers = [
            s for s in self.servers.values() 
            if s.region == region
        ]
        
        if not regional_servers:
            return {}
        
        # Get regional popularity scores
        popularity = self.regional_popularity.get(region, {})
        
        # Sort content by regional popularity
        sorted_content = sorted(
            self.content.values(),
            key=lambda c: popularity.get(c.content_id, 0),
            reverse=True
        )
        
        placement: Dict[str, List[str]] = {
            s.server_id: [] for s in regional_servers
        }
        
        # Tier 1: Popular content goes on ALL servers
        popular_threshold = 0.8  # Top 20% by popularity
        popular_content = sorted_content[:int(len(sorted_content) * 0.2)]
        
        for content in popular_content:
            for server in regional_servers:
                if server.available_bytes >= content.size_bytes:
                    placement[server.server_id].append(content.content_id)
                    server.used_bytes += content.size_bytes
                    server.cached_content.add(content.content_id)
        
        # Tier 2: Long-tail content distributed across servers
        # Spread load with a simple hash-mod (a consistent-hash ring would
        # move far fewer items when servers are added or removed)
        remaining_content = sorted_content[int(len(sorted_content) * 0.2):]
        
        for i, content in enumerate(remaining_content):
            # Assign to server based on hash
            server_idx = hash(content.content_id) % len(regional_servers)
            server = regional_servers[server_idx]
            
            if server.available_bytes >= content.size_bytes:
                placement[server.server_id].append(content.content_id)
                server.used_bytes += content.size_bytes
                server.cached_content.add(content.content_id)
        
        return placement
    
    def select_server_for_request(
        self, 
        content_id: str, 
        user_region: str,
        user_isp: str
    ) -> Optional[OCAServer]:
        """
        Select best OCA server to serve a content request.
        
        Priority:
        1. ISP-embedded OCA with content cached
        2. Regional IXP OCA with content cached
        3. Any OCA with content cached
        4. Fall back to origin (cache miss)
        """
        candidates = []
        
        for server in self.servers.values():
            if content_id not in server.cached_content:
                continue
                
            # Score servers by proximity and load
            score = 0
            
            # Prefer ISP-embedded servers
            if server.location == user_isp:
                score += 100
            # Prefer same-region servers
            elif server.region == user_region:
                score += 50
            
            # Penalize highly loaded servers
            load_pct = server.used_bytes / server.capacity_bytes
            score -= load_pct * 20
            
            candidates.append((score, server))
        
        if not candidates:
            return None  # Cache miss - fetch from origin
        
        # Return highest-scoring server
        candidates.sort(key=lambda x: x[0], reverse=True)
        return candidates[0][1]

# resilience/circuit_breaker.py

"""
Circuit Breaker Implementation
Based on the circuit-breaker pattern popularised by Netflix's Hystrix.
Simplified: it opens after N consecutive failures, whereas Hystrix trips on an
error percentage (default 50%) over a rolling window with a minimum request
volume (default 20 requests in 10 s) and a 5 s sleep window.
"""

from dataclasses import dataclass, field
from typing import Callable, Optional, Any
from enum import Enum
from datetime import datetime, timedelta
import threading
import time


class CircuitState(Enum):
    """Circuit breaker states."""
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if service recovered


@dataclass
class CircuitBreakerConfig:
    """Configuration for circuit breaker behavior."""
    failure_threshold: int = 5         # Failures before opening
    success_threshold: int = 3         # Successes to close from half-open
    timeout_seconds: float = 30.0      # Time before trying half-open
    half_open_max_calls: int = 3       # Max calls in half-open state


@dataclass
class CircuitBreakerStats:
    """Runtime statistics for monitoring."""
    total_calls: int = 0
    successful_calls: int = 0
    failed_calls: int = 0
    rejected_calls: int = 0
    state_changes: int = 0


class CircuitBreaker:
    """
    Circuit Breaker for protecting service calls.
    
    States:
    - CLOSED: Normal operation, requests pass through
    - OPEN: Service failing, reject requests immediately
    - HALF_OPEN: Testing if service recovered
    
    Transitions:
    - CLOSED → OPEN: When failure_threshold exceeded
    - OPEN → HALF_OPEN: After timeout_seconds
    - HALF_OPEN → CLOSED: When success_threshold reached
    - HALF_OPEN → OPEN: On any failure
    
    Applies concepts from:
    - Week 2: Failure handling, timeouts, graceful degradation
    """
    
    def __init__(
        self, 
        name: str, 
        config: CircuitBreakerConfig = None,
        fallback: Callable = None
    ):
        self.name = name
        self.config = config or CircuitBreakerConfig()
        self.fallback = fallback
        
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time: Optional[datetime] = None
        self._half_open_calls = 0
        
        self.stats = CircuitBreakerStats()
        self._lock = threading.Lock()
        
    @property
    def state(self) -> CircuitState:
        """Get current state, checking for timeout transitions."""
        with self._lock:
            if self._state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self._transition_to(CircuitState.HALF_OPEN)
            return self._state
    
    def _should_attempt_reset(self) -> bool:
        """Check if enough time passed to try half-open."""
        if self._last_failure_time is None:
            return False
        elapsed = (datetime.utcnow() - self._last_failure_time).total_seconds()
        return elapsed >= self.config.timeout_seconds
    
    def _transition_to(self, new_state: CircuitState):
        """Transition to new state with logging."""
        old_state = self._state
        self._state = new_state
        self.stats.state_changes += 1
        
        if new_state == CircuitState.HALF_OPEN:
            self._half_open_calls = 0
            self._success_count = 0
        elif new_state == CircuitState.CLOSED:
            self._failure_count = 0
            
        print(f"Circuit {self.name}: {old_state.value} → {new_state.value}")
    
    def call(self, func: Callable, *args, **kwargs) -> Any:
        """
        Execute function through circuit breaker.
        
        Returns result or fallback value.
        Raises exception if no fallback and circuit open.
        """
        self.stats.total_calls += 1
        current_state = self.state
        
        # OPEN: Reject immediately
        if current_state == CircuitState.OPEN:
            self.stats.rejected_calls += 1
            return self._handle_rejection()
        
        # HALF_OPEN: Limit trial calls
        if current_state == CircuitState.HALF_OPEN:
            with self._lock:
                if self._half_open_calls >= self.config.half_open_max_calls:
                    self.stats.rejected_calls += 1
                    return self._handle_rejection()
                self._half_open_calls += 1
        
        # Execute the call
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            return self._handle_failure(e)
    
    def _on_success(self):
        """Handle successful call."""
        with self._lock:
            self.stats.successful_calls += 1
            
            if self._state == CircuitState.HALF_OPEN:
                self._success_count += 1
                if self._success_count >= self.config.success_threshold:
                    self._transition_to(CircuitState.CLOSED)
            elif self._state == CircuitState.CLOSED:
                # Reset failure count on success
                self._failure_count = 0
    
    def _on_failure(self):
        """Handle failed call."""
        with self._lock:
            self.stats.failed_calls += 1
            self._last_failure_time = datetime.utcnow()
            
            if self._state == CircuitState.HALF_OPEN:
                # Any failure in half-open goes back to open
                self._transition_to(CircuitState.OPEN)
            elif self._state == CircuitState.CLOSED:
                self._failure_count += 1
                if self._failure_count >= self.config.failure_threshold:
                    self._transition_to(CircuitState.OPEN)
    
    def _handle_rejection(self) -> Any:
        """Handle rejected call (circuit open)."""
        if self.fallback:
            return self.fallback()
        raise CircuitOpenError(f"Circuit {self.name} is OPEN")
    
    def _handle_failure(self, exception: Exception) -> Any:
        """Handle failed call."""
        if self.fallback:
            return self.fallback()
        raise exception


class CircuitOpenError(Exception):
    """Raised when circuit breaker rejects a call."""
    pass


# Example usage with service calls
class RecommendationService:
    """
    Service that uses circuit breaker for dependency calls.
    """
    
    def __init__(self):
        self.user_service_breaker = CircuitBreaker(
            name="user-service",
            config=CircuitBreakerConfig(
                failure_threshold=5,
                timeout_seconds=30,
                success_threshold=3
            ),
            fallback=self._get_default_user
        )
        
        self.ml_service_breaker = CircuitBreaker(
            name="ml-recommendations",
            config=CircuitBreakerConfig(
                failure_threshold=3,
                timeout_seconds=60,
                success_threshold=2
            ),
            fallback=self._get_popular_content
        )
    
    def get_recommendations(self, user_id: str) -> list:
        """Get personalized recommendations for user."""
        # Get user profile (with circuit breaker)
        user = self.user_service_breaker.call(
            self._fetch_user_profile, user_id
        )
        
        # Get ML recommendations (with circuit breaker)
        recommendations = self.ml_service_breaker.call(
            self._fetch_ml_recommendations, user
        )
        
        return recommendations
    
    def _fetch_user_profile(self, user_id: str) -> dict:
        """Fetch user profile from user service."""
        # Actual HTTP call would go here
        pass
    
    def _fetch_ml_recommendations(self, user: dict) -> list:
        """Fetch recommendations from ML service."""
        # Actual HTTP call would go here
        pass
    
    def _get_default_user(self) -> dict:
        """Fallback: return anonymous user profile."""
        return {"id": "anonymous", "preferences": []}
    
    def _get_popular_content(self) -> list:
        """Fallback: return globally popular content."""
        return [
            {"id": "trending_1", "title": "Popular Movie 1"},
            {"id": "trending_2", "title": "Popular Series 1"},
        ]

# recommendations/recommender.py

"""
Netflix-style Recommendation System
Demonstrates collaborative filtering and personalization.
"""

from dataclasses import dataclass
from typing import Dict, List, Tuple, Optional
import numpy as np
from collections import defaultdict


@dataclass
class UserProfile:
    """User viewing profile."""
    user_id: str
    viewing_history: List[str]  # Content IDs watched
    ratings: Dict[str, float]   # Content ID → rating
    preferences: Dict[str, float]  # Genre → preference score
    

@dataclass
class ContentItem:
    """Content metadata."""
    content_id: str
    title: str
    genres: List[str]
    tags: List[str]
    avg_rating: float
    popularity_score: float


class CollaborativeFilter:
    """
    Collaborative Filtering using Matrix Factorization.
    
    Learns latent factors for users and items that explain
    viewing patterns. Similar users have similar factor vectors.
    
    Applies concepts from:
    - Week 5: Distributed computation (training at scale)
    """
    
    def __init__(self, num_factors: int = 50, learning_rate: float = 0.01):
        self.num_factors = num_factors
        self.learning_rate = learning_rate
        self.user_factors: Dict[str, np.ndarray] = {}
        self.item_factors: Dict[str, np.ndarray] = {}
        
    def fit(self, interactions: List[Tuple[str, str, float]]):
        """
        Train model on user-item interactions.
        
        Args:
            interactions: List of (user_id, item_id, rating) tuples
        """
        # Initialize random factors
        users = set(u for u, _, _ in interactions)
        items = set(i for _, i, _ in interactions)
        
        for user in users:
            self.user_factors[user] = np.random.randn(self.num_factors) * 0.1
        for item in items:
            self.item_factors[item] = np.random.randn(self.num_factors) * 0.1
        
        # Stochastic Gradient Descent
        for epoch in range(20):
            np.random.shuffle(interactions)
            total_error = 0
            
            for user_id, item_id, rating in interactions:
                # Predict rating
                pred = np.dot(
                    self.user_factors[user_id],
                    self.item_factors[item_id]
                )
                error = rating - pred
                total_error += error ** 2
                
                # Update factors
                user_grad = error * self.item_factors[item_id]
                item_grad = error * self.user_factors[user_id]
                
                self.user_factors[user_id] += self.learning_rate * user_grad
                self.item_factors[item_id] += self.learning_rate * item_grad
            
            rmse = np.sqrt(total_error / len(interactions))
            if epoch % 5 == 0:
                print(f"Epoch {epoch}, RMSE: {rmse:.4f}")
    
    def predict(self, user_id: str, item_id: str) -> float:
        """Predict rating for user-item pair."""
        if user_id not in self.user_factors:
            return 3.0  # Default rating for unknown users
        if item_id not in self.item_factors:
            return 3.0  # Default rating for unknown items
            
        return np.dot(
            self.user_factors[user_id],
            self.item_factors[item_id]
        )
    
    def recommend(self, user_id: str, n: int = 10) -> List[Tuple[str, float]]:
        """Get top N recommendations for user."""
        if user_id not in self.user_factors:
            return []
        
        scores = []
        for item_id, item_factors in self.item_factors.items():
            score = np.dot(self.user_factors[user_id], item_factors)
            scores.append((item_id, score))
        
        scores.sort(key=lambda x: x[1], reverse=True)
        return scores[:n]


class ContentBasedFilter:
    """
    Content-Based Filtering using item features.
    
    Recommends items similar to what user has liked,
    based on content features (genre, tags, etc.)
    """
    
    def __init__(self):
        self.item_features: Dict[str, np.ndarray] = {}
        self.feature_index: Dict[str, int] = {}
        
    def index_content(self, items: List[ContentItem]):
        """Build feature vectors for all content."""
        # Build feature vocabulary
        all_features = set()
        for item in items:
            all_features.update(item.genres)
            all_features.update(item.tags)
        
        self.feature_index = {f: i for i, f in enumerate(all_features)}
        num_features = len(self.feature_index)
        
        # Build feature vectors
        for item in items:
            vector = np.zeros(num_features)
            for genre in item.genres:
                vector[self.feature_index[genre]] = 1.0
            for tag in item.tags:
                vector[self.feature_index[tag]] = 0.5
            
            # Normalize
            norm = np.linalg.norm(vector)
            if norm > 0:
                vector /= norm
            
            self.item_features[item.content_id] = vector
    
    def get_similar(self, item_id: str, n: int = 10) -> List[Tuple[str, float]]:
        """Find N most similar items."""
        if item_id not in self.item_features:
            return []
        
        target = self.item_features[item_id]
        similarities = []
        
        for other_id, features in self.item_features.items():
            if other_id == item_id:
                continue
            similarity = np.dot(target, features)
            similarities.append((other_id, similarity))
        
        similarities.sort(key=lambda x: x[1], reverse=True)
        return similarities[:n]
    
    def recommend_for_user(
        self, 
        user: UserProfile, 
        n: int = 10
    ) -> List[Tuple[str, float]]:
        """Recommend based on user's viewing history."""
        # Aggregate scores from similar items to watched content
        scores = defaultdict(float)
        
        for watched_id in user.viewing_history[-20:]:  # Last 20 watched
            similar = self.get_similar(watched_id, n=50)
            for item_id, similarity in similar:
                if item_id not in user.viewing_history:
                    scores[item_id] += similarity
        
        # Sort by aggregated score
        recommendations = sorted(
            scores.items(), 
            key=lambda x: x[1], 
            reverse=True
        )
        return recommendations[:n]


class HybridRecommender:
    """
    Hybrid Recommender combining multiple signals.
    
    Netflix uses ensemble of multiple models:
    - Collaborative filtering
    - Content-based filtering
    - Popularity-based
    - Context-aware (time, device)
    """
    
    def __init__(self):
        self.collaborative = CollaborativeFilter()
        self.content_based = ContentBasedFilter()
        
        # Weights for combining models
        self.weights = {
            "collaborative": 0.4,
            "content_based": 0.3,
            "popularity": 0.2,
            "recency": 0.1
        }
    
    def recommend(
        self, 
        user: UserProfile,
        context: dict = None,
        n: int = 20
    ) -> List[ContentItem]:
        """
        Generate personalized recommendations.
        
        Combines multiple signals with learned weights.
        """
        scores = defaultdict(float)
        
        # Collaborative filtering scores
        cf_recs = self.collaborative.recommend(user.user_id, n=100)
        for item_id, score in cf_recs:
            scores[item_id] += score * self.weights["collaborative"]
        
        # Content-based scores
        cb_recs = self.content_based.recommend_for_user(user, n=100)
        for item_id, score in cb_recs:
            scores[item_id] += score * self.weights["content_based"]
        
        # Filter out already watched
        for watched_id in user.viewing_history:
            scores.pop(watched_id, None)
        
        # Sort and return top N
        sorted_items = sorted(
            scores.items(),
            key=lambda x: x[1],
            reverse=True
        )
        
        return sorted_items[:n]
    
    def get_row_recommendations(
        self,
        user: UserProfile,
        row_type: str
    ) -> List[str]:
        """
        Get recommendations for a specific homepage row.
        
        Row types:
        - "because_you_watched": Similar to recently watched
        - "trending": Popular in user's region
        - "top_picks": Personalized top recommendations
        - "continue_watching": Incomplete titles
        """
        if row_type == "because_you_watched":
            if not user.viewing_history:
                return []
            last_watched = user.viewing_history[-1]
            similar = self.content_based.get_similar(last_watched, n=10)
            return [item_id for item_id, _ in similar]
        
        elif row_type == "top_picks":
            recs = self.recommend(user, n=10)
            return [item_id for item_id, _ in recs]
        
        # ... other row types
        
        return []