Sistema de Recomendación de Películas con Amazon Personalize ​

movie_table = {
    'movie_id': 'string',  # Partition key
    'title': 'string',
    'genres': 'set<string>',
    'year': 'number',
    'director': 'string',
    'cast': 'list<string>',
    'rating': 'number',
    'popularity': 'number',
    'language': 'string',
    'plot': 'string',
    'metadata': 'map'
}

user_table = {
    'user_id': 'string',  # Partition key
    'birth_date': 'string',
    'profession': 'string',
    'favorite_genres': 'list<string>',
    'created_at': 'string',
    'last_active': 'string',
    'preferences': 'map',
    'interaction_count': 'number'
}

# backend/app/services/data_import.py
import pandas as pd
import numpy as np
from datetime import datetime
import boto3
import logging
from typing import List, Dict, Any, Tuple
from pathlib import Path

class MovieDataProcessor:
    def __init__(self, data_path: str):
        self.data_path = Path(data_path)
        self.s3_client = boto3.client('s3')
        self.logger = logging.getLogger(__name__)

    async def process_movie_data(self) -> Tuple[pd.DataFrame, Dict[str, Any]]:
        try:
            # Cargar datasets
            movies_df = pd.read_csv(self.data_path / 'movies.csv')
            ratings_df = pd.read_csv(self.data_path / 'ratings.csv')
            
            # Procesar y limpiar datos
            movies_clean = self._clean_movie_data(movies_df)
            ratings_clean = self._clean_ratings_data(ratings_df)
            
            # Generar metadata adicional
            movies_enriched = self._enrich_movie_data(movies_clean)
            
            # Generar estadísticas
            stats = self._generate_statistics(movies_enriched, ratings_clean)
            
            return movies_enriched, stats
            
        except Exception as e:
            self.logger.error(f"Error processing movie data: {str(e)}")
            raise

    def _clean_movie_data(self, df: pd.DataFrame) -> pd.DataFrame:
        # Eliminar duplicados
        df = df.drop_duplicates(subset=['movieId'])
        
        # Extraer año del título
        df['year'] = df['title'].str.extract(r'\((\d{4})\)').astype(float)
        df['title'] = df['title'].str.replace(r'\s*\(\d{4}\)', '')
        
        # Separar géneros
        df['genres'] = df['genres'].str.split('|')
        
        # Limpiar géneros vacíos
        df['genres'] = df['genres'].apply(
            lambda x: [g.strip() for g in x if g != '(no genres listed)']
        )
        
        # Filtrar películas sin información
        df = df[df['genres'].str.len() > 0]
        
        return df

    def _clean_ratings_data(self, df: pd.DataFrame) -> pd.DataFrame:
        # Convertir timestamp a datetime
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
        
        # Eliminar ratings inválidos
        df = df[df['rating'].between(0.5, 5.0)]
        
        # Ordenar por usuario y timestamp
        df = df.sort_values(['userId', 'timestamp'])
        
        return df

    def _enrich_movie_data(self, df: pd.DataFrame) -> pd.DataFrame:
        # Calcular década
        df['decade'] = (df['year'] // 10) * 10
        
        # Generar categorías por época
        df['era'] = pd.cut(
            df['year'],
            bins=[0, 1950, 1980, 2000, float('inf')],
            labels=['Classic', 'Retro', 'Modern', 'Contemporary']
        )
        
        # Contar géneros por película
        df['genre_count'] = df['genres'].str.len()
        
        return df

    def _generate_statistics(
        self,
        movies_df: pd.DataFrame,
        ratings_df: pd.DataFrame
    ) -> Dict[str, Any]:
        return {
            'total_movies': len(movies_df),
            'total_ratings': len(ratings_df),
            'unique_users': ratings_df['userId'].nunique(),
            'genres_distribution': self._get_genres_distribution(movies_df),
            'ratings_distribution': self._get_ratings_distribution(ratings_df),
            'era_distribution': movies_df['era'].value_counts().to_dict(),
            'processing_date': datetime.now().isoformat()
        }

    def _get_genres_distribution(self, df: pd.DataFrame) -> Dict[str, int]:
        genres_count = {}
        for genres in df['genres']:
            for genre in genres:
                genres_count[genre] = genres_count.get(genre, 0) + 1
        return genres_count

    def _get_ratings_distribution(self, df: pd.DataFrame) -> Dict[float, int]:
        return df['rating'].value_counts().sort_index().to_dict()

# backend/app/services/personalize_data_formatter.py
class PersonalizeDataFormatter:
    def __init__(self):
        self.interaction_schema = {
            'USER_ID': str,
            'ITEM_ID': str,
            'TIMESTAMP': int,
            'EVENT_TYPE': str,
            'EVENT_VALUE': float
        }

        self.item_schema = {
            'ITEM_ID': str,
            'GENRES': str,
            'YEAR': int,
            'ERA': str,
            'GENRE_COUNT': int,
            'CREATION_TIMESTAMP': int
        }

    def format_interactions(
        self,
        ratings_df: pd.DataFrame
    ) -> pd.DataFrame:
        interactions = ratings_df.copy()
        
        # Formatear columnas requeridas
        interactions['USER_ID'] = interactions['userId'].astype(str)
        interactions['ITEM_ID'] = interactions['movieId'].astype(str)
        interactions['TIMESTAMP'] = interactions['timestamp'].astype(np.int64) // 10**9
        interactions['EVENT_TYPE'] = 'RATING'
        interactions['EVENT_VALUE'] = interactions['rating']
        
        # Seleccionar columnas finales
        return interactions[list(self.interaction_schema.keys())]

    def format_items(self, movies_df: pd.DataFrame) -> pd.DataFrame:
        items = movies_df.copy()
        
        # Formatear columnas
        items['ITEM_ID'] = items['movieId'].astype(str)
        items['GENRES'] = items['genres'].apply(lambda x: '|'.join(x))
        items['YEAR'] = items['year'].fillna(-1).astype(int)
        items['ERA'] = items['era']
        items['GENRE_COUNT'] = items['genre_count']
        items['CREATION_TIMESTAMP'] = (
            pd.Timestamp.now().timestamp()
        ).astype(int)
        
        # Seleccionar columnas finales
        return items[list(self.item_schema.keys())]

# backend/app/services/data_uploader.py
class PersonalizeDataUploader:
    def __init__(
        self,
        dataset_group_name: str,
        bucket_name: str,
        region: str
    ):
        self.personalize = boto3.client(
            'personalize',
            region_name=region
        )
        self.s3 = boto3.client('s3', region_name=region)
        self.dataset_group_name = dataset_group_name
        self.bucket_name = bucket_name

    async def upload_and_create_dataset(
        self,
        interactions_df: pd.DataFrame,
        items_df: pd.DataFrame
    ) -> Dict[str, str]:
        try:
            # Subir archivos a S3
            interactions_key = self._upload_to_s3(
                interactions_df,
                'interactions.csv'
            )
            items_key = self._upload_to_s3(
                items_df,
                'items.csv'
            )
            
            # Crear dataset group si no existe
            dataset_group_arn = await self._create_dataset_group()
            
            # Crear schemas
            interaction_schema_arn = await self._create_schema(
                'interactions_schema',
                self._get_interactions_schema()
            )
            
            items_schema_arn = await self._create_schema(
                'items_schema',
                self._get_items_schema()
            )
            
            # Crear datasets
            interactions_ds_arn = await self._create_dataset(
                dataset_group_arn,
                'INTERACTIONS',
                interaction_schema_arn
            )
            
            items_ds_arn = await self._create_dataset(
                dataset_group_arn,
                'ITEMS',
                items_schema_arn
            )
            
            # Importar datos
            await self._import_dataset(
                interactions_ds_arn,
                f's3://{self.bucket_name}/{interactions_key}'
            )
            
            await self._import_dataset(
                items_ds_arn,
                f's3://{self.bucket_name}/{items_key}'
            )
            
            return {
                'dataset_group_arn': dataset_group_arn,
                'interactions_dataset_arn': interactions_ds_arn,
                'items_dataset_arn': items_ds_arn
            }
            
        except Exception as e:
            self.logger.error(f"Error uploading data: {str(e)}")
            raise

    def _upload_to_s3(
        self,
        df: pd.DataFrame,
        filename: str
    ) -> str:
        csv_buffer = io.StringIO()
        df.to_csv(csv_buffer, index=False)
        
        key = f'personalize/{filename}'
        self.s3.put_object(
            Bucket=self.bucket_name,
            Key=key,
            Body=csv_buffer.getvalue()
        )
        
        return key

    async def _create_dataset_group(self) -> str:
        response = self.personalize.create_dataset_group(
            name=self.dataset_group_name
        )
        
        dataset_group_arn = response['datasetGroupArn']
        
        # Esperar a que esté activo
        waiter = self.personalize.get_waiter('dataset_group_available')
        waiter.wait(
            datasetGroupArn=dataset_group_arn,
            WaiterConfig={
                'Delay': 30,
                'MaxAttempts': 60
            }
        )
        
        return dataset_group_arn

    async def _create_schema(
        self,
        name: str,
        schema: Dict[str, Any]
    ) -> str:
        response = self.personalize.create_schema(
            name=name,
            schema=json.dumps(schema)
        )
        return response['schemaArn']

    async def _create_dataset(
        self,
        dataset_group_arn: str,
        dataset_type: str,
        schema_arn: str
    ) -> str:
        response = self.personalize.create_dataset(
            name=f"{self.dataset_group_name}_{dataset_type.lower()}",
            schemaArn=schema_arn,
            datasetGroupArn=dataset_group_arn,
            datasetType=dataset_type
        )
        return response['datasetArn']

    async def _import_dataset(
        self,
        dataset_arn: str,
        data_location: str
    ) -> None:
        response = self.personalize.create_dataset_import_job(
            jobName=f"import_{int(time.time())}",
            datasetArn=dataset_arn,
            dataSource={'dataLocation': data_location},
            roleArn=self.role_arn
        )
        
        # Esperar a que termine la importación
        import_job_arn = response['datasetImportJobArn']
        waiter = self.personalize.get_waiter('dataset_import_job_available')
        waiter.wait(
            datasetImportJobArn=import_job_arn,
            WaiterConfig={
                'Delay': 60,
                'MaxAttempts': 60
            }
        )

    def _get_interactions_schema(self) -> Dict[str, Any]:
        return {
            "type": "record",
            "name": "Interactions",
            "namespace": "com.movie.recommender",
            "fields": [
                {"name": "USER_ID", "type": "string"},
                {"name": "ITEM_ID", "type": "string"},
                {"name": "TIMESTAMP", "type": "long"},
                {"name": "EVENT_TYPE", "type": "string"},
                {"name": "EVENT_VALUE", "type": "float"}
            ]
        }

    def _get_items_schema(self) -> Dict[str, Any]:
        return {
            "type": "record",
            "name": "Items",
            "namespace": "com.movie.recommender",
            "fields": [
                {"name": "ITEM_ID", "type": "string"},
                {"name": "GENRES", "type": "string"},
                {"name": "YEAR", "type": "int"},
                {"name": "ERA", "type": "string"},
                {"name": "GENRE_COUNT", "type": "int"},
                {"name": "CREATION_TIMESTAMP", "type": "long"}
            ]
        }

# backend/app/services/personalize_manager.py
import boto3
import json
import logging
from typing import Dict, Any, List, Optional
from datetime import datetime
import time

class PersonalizeManager:
    def __init__(self, region: str, role_arn: str):
        self.personalize = boto3.client('personalize', region_name=region)
        self.personalize_runtime = boto3.client('personalize-runtime', region_name=region)
        self.role_arn = role_arn
        self.logger = logging.getLogger(__name__)

    async def create_solution(
        self,
        dataset_group_arn: str,
        recipe_arn: str,
        solution_name: str
    ) -> Dict[str, Any]:
        try:
            # Crear solución
            create_solution_response = self.personalize.create_solution(
                name=solution_name,
                datasetGroupArn=dataset_group_arn,
                recipeArn=recipe_arn,
                performAutoML=False,
                performHPO=True,  # Optimización de hiperparámetros
                solutionConfig={
                    "algorithmHyperParameters": {
                        "hidden_dimension": "64",
                        "bptt": "32",
                        "recency_mask": "true",
                        "learning_rate": "0.001",
                        "dropout": "0.5"
                    },
                    "featureTransformationParameters": {
                        "genres_column": "GENRES",
                        "year_column": "YEAR"
                    },
                    "eventValueThreshold": "0.5"
                }
            )
            
            solution_arn = create_solution_response['solutionArn']
            
            # Crear versión de solución (entrenar)
            create_solution_version_response = self.personalize.create_solution_version(
                solutionArn=solution_arn
            )
            
            solution_version_arn = create_solution_version_response['solutionVersionArn']
            
            # Esperar a que termine el entrenamiento
            await self._wait_for_solution_version(solution_version_arn)
            
            # Obtener métricas
            metrics = await self._get_solution_metrics(solution_version_arn)
            
            return {
                'solution_arn': solution_arn,
                'solution_version_arn': solution_version_arn,
                'metrics': metrics
            }
            
        except Exception as e:
            self.logger.error(f"Error creating solution: {str(e)}")
            raise

    async def create_campaign(
        self,
        solution_version_arn: str,
        campaign_name: str,
        min_provisioned_tps: int = 1
    ) -> str:
        try:
            response = self.personalize.create_campaign(
                name=campaign_name,
                solutionVersionArn=solution_version_arn,
                minProvisionedTPS=min_provisioned_tps,
                campaignConfig={
                    "itemExplorationConfig": {
                        "explorationWeight": "0.3",
                        "explorationItemAgeCutOff": "30"
                    }
                }
            )
            
            campaign_arn = response['campaignArn']
            
            # Esperar a que la campaña esté activa
            waiter = self.personalize.get_waiter('campaign_available')
            waiter.wait(
                campaignArn=campaign_arn,
                WaiterConfig={
                    'Delay': 60,
                    'MaxAttempts': 60
                }
            )
            
            return campaign_arn
            
        except Exception as e:
            self.logger.error(f"Error creating campaign: {str(e)}")
            raise

    async def create_filter(
        self,
        dataset_group_arn: str,
        filter_name: str
    ) -> str:
        try:
            # Crear filtro para excluir items ya vistos
            response = self.personalize.create_filter(
                name=filter_name,
                datasetGroupArn=dataset_group_arn,
                filterExpression="EXCLUDE ItemID WHERE INTERACTIONS.event_type IN ('RATING', 'WATCH')"
            )
            
            return response['filterArn']
            
        except Exception as e:
            self.logger.error(f"Error creating filter: {str(e)}")
            raise

    async def get_recommendations(
        self,
        campaign_arn: str,
        user_id: str,
        num_results: int = 10,
        filter_arn: Optional[str] = None,
        context: Optional[Dict[str, str]] = None
    ) -> List[Dict[str, Any]]:
        try:
            params = {
                'campaignArn': campaign_arn,
                'userId': user_id,
                'numResults': num_results
            }
            
            if filter_arn:
                params['filterArn'] = filter_arn
            
            if context:
                params['context'] = context
            
            response = self.personalize_runtime.get_recommendations(**params)
            
            return response['itemList']
            
        except Exception as e:
            self.logger.error(f"Error getting recommendations: {str(e)}")
            raise

    async def _wait_for_solution_version(
        self,
        solution_version_arn: str
    ) -> None:
        while True:
            response = self.personalize.describe_solution_version(
                solutionVersionArn=solution_version_arn
            )
            status = response['solutionVersion']['status']
            
            if status == 'ACTIVE':
                break
            elif status == 'CREATE FAILED':
                raise Exception(
                    f"Solution version creation failed: {response['solutionVersion']['failureReason']}"
                )
                
            time.sleep(60)

    async def _get_solution_metrics(
        self,
        solution_version_arn: str
    ) -> Dict[str, float]:
        response = self.personalize.get_solution_metrics(
            solutionVersionArn=solution_version_arn
        )
        return response['metrics']

# backend/app/services/personalize_trainer.py
class PersonalizeTrainer:
    def __init__(
        self,
        personalize_manager: PersonalizeManager,
        config: Dict[str, Any]
    ):
        self.manager = personalize_manager
        self.config = config
        self.logger = logging.getLogger(__name__)

    async def train_models(
        self,
        dataset_group_arn: str
    ) -> Dict[str, Any]:
        try:
            # Crear solución user-personalization
            user_personalization = await self.manager.create_solution(
                dataset_group_arn=dataset_group_arn,
                recipe_arn='arn:aws:personalize:::recipe/aws-user-personalization',
                solution_name=f"{self.config['project_name']}_user_personalization"
            )
            
            # Crear solución similar-items
            similar_items = await self.manager.create_solution(
                dataset_group_arn=dataset_group_arn,
                recipe_arn='arn:aws:personalize:::recipe/aws-sims',
                solution_name=f"{self.config['project_name']}_similar_items"
            )
            
            # Crear campañas
            user_campaign_arn = await self.manager.create_campaign(
                solution_version_arn=user_personalization['solution_version_arn'],
                campaign_name=f"{self.config['project_name']}_user_campaign"
            )
            
            similar_items_campaign_arn = await self.manager.create_campaign(
                solution_version_arn=similar_items['solution_version_arn'],
                campaign_name=f"{self.config['project_name']}_similar_items_campaign"
            )
            
            # Crear filtros
            filter_arn = await self.manager.create_filter(
                dataset_group_arn=dataset_group_arn,
                filter_name=f"{self.config['project_name']}_viewed_filter"
            )
            
            return {
                'user_personalization': {
                    'solution_arn': user_personalization['solution_arn'],
                    'solution_version_arn': user_personalization['solution_version_arn'],
                    'campaign_arn': user_campaign_arn,
                    'metrics': user_personalization['metrics']
                },
                'similar_items': {
                    'solution_arn': similar_items['solution_arn'],
                    'solution_version_arn': similar_items['solution_version_arn'],
                    'campaign_arn': similar_items_campaign_arn,
                    'metrics': similar_items['metrics']
                },
                'filter_arn': filter_arn
            }
            
        except Exception as e:
            self.logger.error(f"Error training models: {str(e)}")
            raise

# backend/app/services/personalize_monitor.py
class PersonalizeMonitor:
    def __init__(self, cloudwatch_client):
        self.cloudwatch = cloudwatch_client
        self.logger = logging.getLogger(__name__)

    async def monitor_solution_metrics(
        self,
        solution_version_arn: str
    ) -> Dict[str, Any]:
        try:
            metrics = {}
            
            # Métricas de precisión
            metrics['precision'] = await self._get_metric(
                solution_version_arn,
                'precision_at_k'
            )
            
            # Métricas de recall
            metrics['recall'] = await self._get_metric(
                solution_version_arn,
                'recall_at_k'
            )
            
            # Normalized Discounted Cumulative Gain
            metrics['ndcg'] = await self._get_metric(
                solution_version_arn,
                'ndcg_at_k'
            )
            
            return metrics
            
        except Exception as e:
            self.logger.error(f"Error monitoring metrics: {str(e)}")
            raise

    async def create_performance_dashboard(
        self,
        campaign_arn: str,
        solution_version_arn: str
    ) -> str:
        try:
            dashboard_name = f"personalize-metrics-{int(time.time())}"
            
            dashboard_body = {
                "widgets": [
                    {
                        "type": "metric",
                        "properties": {
                            "metrics": [
                                ["AWS/Personalize", "GetRecommendations.Latency", "CampaignArn", campaign_arn],
                                [".", "GetRecommendations.RequestCount", ".", "."]
                            ],
                            "period": 300,
                            "stat": "Average",
                            "region": self.cloudwatch.meta.region_name,
                            "title": "Campaign Performance"
                        }
                    },
                    {
                        "type": "metric",
                        "properties": {
                            "metrics": [
                                ["AWS/Personalize", "precision_at_k", "SolutionVersionArn", solution_version_arn],
                                [".", "recall_at_k", ".", "."],
                                [".", "ndcg_at_k", ".", "."]
                            ],
                            "period": 3600,
                            "stat": "Average",
                            "region": self.cloudwatch.meta.region_name,
                            "title": "Model Metrics"
                        }
                    }
                ]
            }
            
            self.cloudwatch.put_dashboard(
                DashboardName=dashboard_name,
                DashboardBody=json.dumps(dashboard_body)
            )
            
            return dashboard_name
            
        except Exception as e:
            self.logger.error(f"Error creating dashboard: {str(e)}")
            raise

    async def _get_metric(
        self,
        resource_arn: str,
        metric_name: str
    ) -> float:
        response = self.cloudwatch.get_metric_statistics(
            Namespace='AWS/Personalize',
            MetricName=metric_name,
            Dimensions=[
                {
                    'Name': 'SolutionVersionArn',
                    'Value': resource_arn
                }
            ],
            StartTime=datetime.utcnow() - timedelta(hours=1),
            EndTime=datetime.utcnow(),
            Period=3600,
            Statistics=['Average']
        )
        
        if response['Datapoints']:
            return response['Datapoints'][0]['Average']
        return 0.0

# backend/app/api/main.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from mangum import Mangum
from app.core.config import Settings
from app.api.routes import user, movie, recommendations
from app.core.logging import setup_logging

app = FastAPI(
    title="Movie Recommender API",
    description="API for personalized movie recommendations",
    version="1.0.0"
)

# Configurar CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Rutas
app.include_router(user.router, prefix="/api/users", tags=["users"])
app.include_router(movie.router, prefix="/api/movies", tags=["movies"])
app.include_router(recommendations.router, prefix="/api/recommendations", tags=["recommendations"])

# Handler para Lambda
handler = Mangum(app)

# backend/app/models/user.py
from pydantic import BaseModel, Field, validator
from typing import List, Optional
from datetime import date
import re

class UserBase(BaseModel):
    birth_date: date = Field(..., description="User's date of birth")
    profession: str = Field(..., description="User's profession")
    favorite_genres: List[str] = Field(..., min_items=1, max_items=3, description="List of favorite movie genres")

    @validator('profession')
    def validate_profession(cls, v):
        allowed_professions = [
            "student", "engineer", "doctor", "artist", "teacher",
            "business", "technology", "healthcare", "other"
        ]
        if v.lower() not in allowed_professions:
            raise ValueError(f"Profession must be one of: {', '.join(allowed_professions)}")
        return v.lower()

    @validator('favorite_genres')
    def validate_genres(cls, v):
        allowed_genres = [
            "action", "comedy", "drama", "horror", "sci-fi",
            "thriller", "romance", "documentary", "animation"
        ]
        for genre in v:
            if genre.lower() not in allowed_genres:
                raise ValueError(f"Genre must be one of: {', '.join(allowed_genres)}")
        return [genre.lower() for genre in v]

class UserCreate(UserBase):
    pass

class UserUpdate(UserBase):
    birth_date: Optional[date] = None
    profession: Optional[str] = None
    favorite_genres: Optional[List[str]] = None

class UserInDB(UserBase):
    id: str
    created_at: date
    last_active: date
    recommendation_count: int = 0

    class Config:
        orm_mode = True

# backend/app/models/movie.py
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import date

class MovieBase(BaseModel):
    title: str = Field(..., min_length=1)
    year: int = Field(..., ge=1900, le=date.today().year)
    genres: List[str]
    director: Optional[str] = None
    cast: Optional[List[str]] = None
    plot: Optional[str] = None
    rating: Optional[float] = Field(None, ge=0, le=10)
    duration: Optional[int] = Field(None, ge=0)  # en minutos

class MovieCreate(MovieBase):
    pass

class MovieUpdate(MovieBase):
    pass

class MovieInDB(MovieBase):
    id: str
    created_at: date
    updated_at: date
    recommendation_count: int = 0

    class Config:
        orm_mode = True

# backend/app/models/recommendation.py
from pydantic import BaseModel, Field
from typing import List, Dict
from datetime import datetime

class RecommendationBase(BaseModel):
    user_id: str
    movie_ids: List[str]
    score: float = Field(..., ge=0, le=1)
    context: Dict[str, str] = {}

class RecommendationCreate(RecommendationBase):
    pass

class RecommendationInDB(RecommendationBase):
    id: str
    created_at: datetime
    accepted: Optional[bool] = None
    feedback: Optional[str] = None

    class Config:
        orm_mode = True

# backend/app/db/dynamodb.py
import boto3
from botocore.exceptions import ClientError
from typing import Dict, Any, Optional, List
import logging
import json
from datetime import datetime

class DynamoDBManager:
    def __init__(self, region: str):
        self.dynamodb = boto3.resource('dynamodb', region_name=region)
        self.logger = logging.getLogger(__name__)

    async def get_item(
        self,
        table_name: str,
        key: Dict[str, str]
    ) -> Optional[Dict[str, Any]]:
        try:
            table = self.dynamodb.Table(table_name)
            response = table.get_item(Key=key)
            return response.get('Item')
        except ClientError as e:
            self.logger.error(f"Error getting item: {str(e)}")
            raise

    async def put_item(
        self,
        table_name: str,
        item: Dict[str, Any]
    ) -> None:
        try:
            table = self.dynamodb.Table(table_name)
            table.put_item(Item=item)
        except ClientError as e:
            self.logger.error(f"Error putting item: {str(e)}")
            raise

    async def query_items(
        self,
        table_name: str,
        key_condition_expression: str,
        expression_attribute_values: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        try:
            table = self.dynamodb.Table(table_name)
            response = table.query(
                KeyConditionExpression=key_condition_expression,
                ExpressionAttributeValues=expression_attribute_values
            )
            return response.get('Items', [])
        except ClientError as e:
            self.logger.error(f"Error querying items: {str(e)}")
            raise

    async def update_item(
        self,
        table_name: str,
        key: Dict[str, str],
        update_expression: str,
        expression_attribute_values: Dict[str, Any]
    ) -> None:
        try:
            table = self.dynamodb.Table(table_name)
            table.update_item(
                Key=key,
                UpdateExpression=update_expression,
                ExpressionAttributeValues=expression_attribute_values
            )
        except ClientError as e:
            self.logger.error(f"Error updating item: {str(e)}")
            raise

# backend/app/repositories/user_repository.py
from typing import Optional, List
from app.models.user import UserCreate, UserInDB
from app.db.dynamodb import DynamoDBManager
import uuid
from datetime import datetime

class UserRepository:
    def __init__(self, db_manager: DynamoDBManager):
        self.db = db_manager
        self.table_name = "users"

    async def create_user(self, user: UserCreate) -> UserInDB:
        user_id = str(uuid.uuid4())
        now = datetime.utcnow()
        
        user_data = {
            "id": user_id,
            "birth_date": user.birth_date.isoformat(),
            "profession": user.profession,
            "favorite_genres": user.favorite_genres,
            "created_at": now.isoformat(),
            "last_active": now.isoformat(),
            "recommendation_count": 0
        }
        
        await self.db.put_item(self.table_name, user_data)
        return UserInDB(**user_data)

    async def get_user(self, user_id: str) -> Optional[UserInDB]:
        user_data = await self.db.get_item(
            self.table_name,
            {"id": user_id}
        )
        
        if user_data:
            return UserInDB(**user_data)
        return None

    async def update_user(
        self,
        user_id: str,
        update_data: Dict[str, Any]
    ) -> Optional[UserInDB]:
        update_expr_parts = []
        expr_attr_values = {}
        
        for key, value in update_data.items():
            if value is not None:
                update_expr_parts.append(f"#{key} = :{key}")
                expr_attr_values[f":{key}"] = value
        
        if not update_expr_parts:
            return None
            
        update_expression = "SET " + ", ".join(update_expr_parts)
        
        await self.db.update_item(
            self.table_name,
            {"id": user_id},
            update_expression,
            expr_attr_values
        )
        
        return await self.get_user(user_id)

    async def increment_recommendation_count(
        self,
        user_id: str
    ) -> None:
        await self.db.update_item(
            self.table_name,
            {"id": user_id},
            "SET recommendation_count = recommendation_count + :inc",
            {":inc": 1}
        )

# backend/app/repositories/movie_repository.py
class MovieRepository:
    def __init__(self, db_manager: DynamoDBManager):
        self.db = db_manager
        self.table_name = "movies"

    async def get_movies_by_ids(
        self,
        movie_ids: List[str]
    ) -> List[MovieInDB]:
        movies = []
        for movie_id in movie_ids:
            movie_data = await self.db.get_item(
                self.table_name,
                {"id": movie_id}
            )
            if movie_data:
                movies.append(MovieInDB(**movie_data))
        return movies

    async def get_movies_by_genre(
        self,
        genre: str,
        limit: int = 10
    ) -> List[MovieInDB]:
        movies_data = await self.db.query_items(
            self.table_name,
            "contains(genres, :genre)",
            {":genre": genre}
        )
        return [MovieInDB(**m) for m in movies_data[:limit]]

# backend/app/repositories/recommendation_repository.py
class RecommendationRepository:
    def __init__(self, db_manager: DynamoDBManager):
        self.db = db_manager
        self.table_name = "recommendations"

    async def create_recommendation(
        self,
        recommendation: RecommendationCreate
    ) -> RecommendationInDB:
        recommendation_id = str(uuid.uuid4())
        now = datetime.utcnow()
        
        recommendation_data = {
            "id": recommendation_id,
            "user_id": recommendation.user_id,
            "movie_ids": recommendation.movie_ids,
            "score": recommendation.score,
            "context": recommendation.context,
            "created_at": now.isoformat(),
            "accepted": None,
            "feedback": None
        }
        
        await self.db.put_item(self.table_name, recommendation_data)
        return RecommendationInDB(**recommendation_data)

    async def get_user_recommendations(
        self,
        user_id: str,
        limit: int = 10
    ) -> List[RecommendationInDB]:
        recommendations_data = await self.db.query_items(
            self.table_name,
            "user_id = :user_id",
            {":user_id": user_id}
        )
        
        recommendations = [
            RecommendationInDB(**r) for r in recommendations_data
        ]
        
        return sorted(
            recommendations,
            key=lambda x: x.created_at,
            reverse=True
        )[:limit]

    async def update_recommendation_feedback(
        self,
        recommendation_id: str,
        accepted: bool,
        feedback: Optional[str] = None
    ) -> None:
        update_expression = "SET accepted = :accepted"
        expression_values = {":accepted": accepted}
        
        if feedback:
            update_expression += ", feedback = :feedback"
            expression_values[":feedback"] = feedback
        
        await self.db.update_item(
            self.table_name,
            {"id": recommendation_id},
            update_expression,
            expression_values
        )