New feed recommender

app.py

@@ -37,7 +37,6 @@ SEED = 42
 random.seed(SEED)
 np.random.seed(SEED)
 
-CACHE_SIZE = 1000
 TOP_K = 75
 HISTORY_WINDOW = timedelta(days=30)
 TIMEZONE = ZoneInfo("UTC")
@@ -48,7 +47,7 @@ supabase_client = None
 user_interactions = defaultdict(set)
 post_features = {}
 post_metadata = {}
-recommendation_cache = {}
+
 if not os.path.exists('/tmp/cache'):
     os.makedirs('/tmp/cache')
 sentence_model = SentenceTransformer('all-MiniLM-L6-v2', cache_folder='/tmp/cache')
@@ -56,100 +55,47 @@ sentence_model = SentenceTransformer('all-MiniLM-L6-v2', cache_folder='/tmp/cach
 SUPABASE_URL = os.getenv('supabaseUrl')
 SUPABASE_KEY = os.getenv('supabaseAnonKey')
 
-
 def get_supabase_client():
     global supabase_client
     if supabase_client is None:
         supabase_client = create_client(SUPABASE_URL, SUPABASE_KEY)
     return supabase_client
 
-
 def parse_datetime(dt_str: str) -> datetime:
-    """Parse ISO datetime string and ensure correct microsecond precision."""
     try:
-        if '.' in dt_str:
-            date_part, time_part = dt_str.split('T')
-            time_part, tz_part = time_part.split('+')
-            if '.' in time_part:
-                time_without_micro, micro = time_part.split('.')
-                micro = micro.ljust(6, '0')  # Ensure microseconds are 6 digits
-                time_part = f"{time_without_micro}.{micro}"
-            dt_str = f"{date_part}T{time_part}+00:00"
         return datetime.fromisoformat(dt_str).astimezone(TIMEZONE)
     except Exception as e:
         logger.error(f"Error parsing datetime: {dt_str} - {str(e)}")
         raise
 
-
 class Recommender:
     def __init__(self):
         self.user_profiles = defaultdict(lambda: np.zeros(384))
         self.post_popularity = defaultdict(float)
         self.last_update = datetime.now(TIMEZONE) - HISTORY_WINDOW
 
-    async def update_interactions(self):
+    async def update_data(self):
         supabase = get_supabase_client()
 
-        # Fetch new likes
-        likes = await asyncio.to_thread(
-            supabase.table('likes')
-            .select('user_id, post_id, created_at')
-            .gt('created_at', self.last_update.isoformat())
-            .execute
-        )
-
-        # Fetch new comments
-        comments = await asyncio.to_thread(
-            supabase.table('comments')
-            .select('user_id, post_id, created_at')
-            .gt('created_at', self.last_update.isoformat())
-            .execute
-        )
-
-        for like in likes.data:
-            user_id = like['user_id']
-            post_id = like['post_id']
-            timestamp = parse_datetime(like['created_at'])
-            user_interactions[user_id].add(post_id)
-            self.post_popularity[post_id] += 1.0
-            if post_id in post_features:
-                self.user_profiles[user_id] = (
-                        0.5 * self.user_profiles[user_id] +
-                        0.5 * post_features[post_id]
-                )
-
-        for comment in comments.data:
-            user_id = comment['user_id']
-            post_id = comment['post_id']
-            timestamp = parse_datetime(comment['created_at'])
+        # Fetch likes and comments
+        likes = await asyncio.to_thread(supabase.table('likes').select('user_id, post_id, created_at').gt('created_at', self.last_update.isoformat()).execute)
+        comments = await asyncio.to_thread(supabase.table('comments').select('user_id, post_id, created_at').gt('created_at', self.last_update.isoformat()).execute)
+
+        for interaction in likes.data + comments.data:
+            user_id = interaction['user_id']
+            post_id = interaction['post_id']
             user_interactions[user_id].add(post_id)
-            self.post_popularity[post_id] += 0.5
+            self.post_popularity[post_id] += 1.0 if 'likes' else 0.5
             if post_id in post_features:
-                self.user_profiles[user_id] = (
-                        0.5 * self.user_profiles[user_id] +
-                        0.5 * post_features[post_id]
-                )
+                self.user_profiles[user_id] += post_features[post_id]
 
-        self.last_update = datetime.now(TIMEZONE)
-        logger.info(f"Updated interactions: {len(likes.data)} new likes, {len(comments.data)} new comments")
+        # Fetch posts
+        posts = await asyncio.to_thread(supabase.table('posts').select('*').gt('created_at', self.last_update.isoformat()).execute)
 
-    async def update_posts(self):
-        supabase = get_supabase_client()
-
-        posts = await asyncio.to_thread(
-            supabase.table('posts')
-            .select('*')
-            .gt('created_at', self.last_update.isoformat())
-            .execute
-        )
-
-        post_texts = []
-        post_ids = []
+        post_texts, post_ids = [], []
         for post in posts.data:
             post_id = post['id']
-            title = post.get('movie_name', '') or ''
-            description = post.get('content', '') or ''
-            text = f"{title} {description}".strip()
+            text = f"{post.get('movie_name', '')} {post.get('content', '')}".strip()
             post_texts.append(text)
             post_ids.append(post_id)
             post['created_at'] = parse_datetime(post['created_at'])
@@ -161,16 +107,10 @@ class Recommender:
             for post_id, embedding in zip(post_ids, embeddings):
                 post_features[post_id] = embedding / np.linalg.norm(embedding)
 
-        logger.info(f"Updated {len(posts.data)} posts")
-
-    async def full_update(self):
-        await self.update_posts()
-        await self.update_interactions()
+        self.last_update = datetime.now(TIMEZONE)
+        logger.info("Data updated: {} posts, {} interactions".format(len(posts.data), len(likes.data) + len(comments.data)))
 
     def get_recommendations(self, user_id: str) -> List[Dict]:
-        if user_id in recommendation_cache:
-            return recommendation_cache[user_id]
-
         user_profile = self.user_profiles[user_id]
         seen_posts = user_interactions[user_id]
 
@@ -184,19 +124,13 @@ class Recommender:
             sim_score = np.dot(user_profile, feature) if np.any(user_profile) else 0
             time_diff = now - post_metadata[post_id]['created_at']
             freshness = 1.0 / (1.0 + (time_diff.days / 7))
-            score = (
-                    0.6 * sim_score +
-                    0.3 * np.log1p(self.post_popularity[post_id]) +
-                    0.1 * freshness
-            )
+
+            score = 0.6 * sim_score + 0.3 * np.log1p(self.post_popularity[post_id]) + 0.1 * freshness
             scores[post_id] = score + random.uniform(-0.1, 0.1)
 
         top_posts = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:TOP_K]
         results = [post_metadata[post_id] for post_id, _ in top_posts]
-
-        recommendation_cache[user_id] = results
-        if len(recommendation_cache) > CACHE_SIZE:
-            recommendation_cache.pop(next(iter(recommendation_cache)))
+        random.shuffle(results)
 
         return results
 
@@ -204,8 +138,7 @@ recommender = Recommender()
 scheduler = BackgroundScheduler(timezone="UTC")
 
 async def background_update():
-    await recommender.update_posts()
-    await recommender.update_interactions()
+    await recommender.update_data()
 
 def sync_background_update():
     loop = asyncio.new_event_loop()
@@ -218,8 +151,8 @@ async def get_recommendations_handler(user_id: str = Query(...)):
     try:
         recommendations = recommender.get_recommendations(user_id)
 
-        if recommendations:
-            insert_pos = random.randint(0, min(9, len(recommendations)-1))
+        if recommendations and not any(item.get("type") == "suggestedaccounts" for item in recommendations):
+            insert_pos = random.randint(0, min(9, len(recommendations) - 1))
             recommendations.insert(insert_pos, {"type": "suggestedaccounts"})
 
         return {"status": "success", "recommendations": recommendations}
@@ -229,10 +162,9 @@ async def get_recommendations_handler(user_id: str = Query(...)):
 
 @app.on_event("startup")
 async def startup_event():
-    await recommender.full_update()
+    await recommender.update_data()
     scheduler.add_job(sync_background_update, 'interval', seconds=UPDATE_INTERVAL)
     scheduler.start()
-    logger.info(f"Recommender initialized with background updates every {UPDATE_INTERVAL} seconds")
 
 @app.on_event("shutdown")
 async def shutdown_event():
@@ -242,7 +174,3 @@ async def shutdown_event():
 @app.get("/")
 async def health_check():
     return {"status": "success", "message": "Service operational"}
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)