New feed recommender

app.py

@@ -18,7 +18,10 @@ from apscheduler.schedulers.background import BackgroundScheduler
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
+# Load environment variables
 load_dotenv()
+
+# FastAPI app setup
 app = FastAPI()
 
 app.add_middleware(
@@ -38,7 +41,7 @@ CACHE_SIZE = 1000
 TOP_K = 75
 HISTORY_WINDOW = timedelta(days=30)
 TIMEZONE = ZoneInfo("UTC")
-UPDATE_INTERVAL = 300
+UPDATE_INTERVAL = 300  # In seconds (5 minutes)
 
 # Global variables
 supabase_client = None
@@ -53,36 +56,30 @@ 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
 
-async def fetch_full_post_records(post_ids: List[str], batch_size=1000):
-    supabase = get_supabase_client()
-    if not post_ids:
-        return []
-
-    records = []
-    tasks = []
-    for i in range(0, len(post_ids), batch_size):
-        batch_ids = post_ids[i:i + batch_size]
-        task = asyncio.to_thread(
-            supabase.table('posts').select('*').in_('id', batch_ids).execute
-        )
-        tasks.append(task)
-
-    responses = await asyncio.gather(*tasks)
-    for response in responses:
-        batch_records = response.data
-        for record in batch_records:
-            record['type'] = 'post'
-            record['created_at'] = datetime.fromisoformat(
-                record['created_at'].replace('Z', '+00:00')
-            ).astimezone(TIMEZONE)
-        records.extend(batch_records)
-    return records
+
+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):
@@ -92,45 +89,45 @@ class Recommender:
 
     async def update_interactions(self):
         supabase = get_supabase_client()
-        last_update = self.last_update
 
         # Fetch new likes
         likes = await asyncio.to_thread(
             supabase.table('likes')
             .select('user_id, post_id, created_at')
-            .gt('created_at', last_update.isoformat())
+            .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', last_update.isoformat())
+            .gt('created_at', self.last_update.isoformat())
             .execute
         )
 
         for like in likes.data:
             user_id = like['user_id']
             post_id = like['post_id']
-            timestamp = datetime.fromisoformat(like['created_at'].replace('Z', '+00:00')).astimezone(TIMEZONE)
+            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]
+                        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 = datetime.fromisoformat(comment['created_at'].replace('Z', '+00:00')).astimezone(TIMEZONE)
+            timestamp = parse_datetime(comment['created_at'])
             user_interactions[user_id].add(post_id)
             self.post_popularity[post_id] += 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]
+                        0.5 * self.user_profiles[user_id] +
+                        0.5 * post_features[post_id]
                 )
 
         self.last_update = datetime.now(TIMEZONE)
@@ -138,7 +135,7 @@ class Recommender:
 
     async def update_posts(self):
         supabase = get_supabase_client()
-        # Fetch new or updated posts since last update
+
         posts = await asyncio.to_thread(
             supabase.table('posts')
             .select('*')
@@ -155,9 +152,7 @@ class Recommender:
             text = f"{title} {description}".strip()
             post_texts.append(text)
             post_ids.append(post_id)
-            post['created_at'] = datetime.fromisoformat(
-                post['created_at'].replace('Z', '+00:00')
-            ).astimezone(TIMEZONE)
+            post['created_at'] = parse_datetime(post['created_at'])
             post['type'] = 'post'
             post_metadata[post_id] = post
 
@@ -169,49 +164,12 @@ class Recommender:
         logger.info(f"Updated {len(posts.data)} posts")
 
     async def full_update(self):
-        """Full refresh at startup"""
-        supabase = get_supabase_client()
-        posts = await asyncio.to_thread(
-            supabase.table('posts')
-            .select('*')
-            .execute
-        )
-
-        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()
-            post_texts.append(text)
-            post_ids.append(post_id)
-
-            # Adjust the timestamp to ensure six digits for microseconds
-            created_at_str = post['created_at']
-            if '.' in created_at_str:
-                date_part, time_part = created_at_str.split('T')
-                time_part = time_part.split('+')[0]  # Remove timezone for now
-                if '.' in time_part:
-                    time_without_micro, micro = time_part.split('.')
-                    micro = micro.zfill(6)  # Pad with zeros to make it six digits
-                    time_part = f"{time_without_micro}.{micro}"
-                created_at_str = f"{date_part}T{time_part}+00:00"
-            post['created_at'] = datetime.fromisoformat(created_at_str).astimezone(TIMEZONE)
-            post['type'] = 'post'
-            post_metadata[post_id] = post
-
-        if post_texts:
-            embeddings = sentence_model.encode(post_texts, show_progress_bar=True, convert_to_numpy=True)
-            for post_id, embedding in zip(post_ids, embeddings):
-                post_features[post_id] = embedding / np.linalg.norm(embedding)
-
+        await self.update_posts()
         await self.update_interactions()
 
     def get_recommendations(self, user_id: str) -> List[Dict]:
         if user_id in recommendation_cache:
-            cached = recommendation_cache[user_id]
-            return random.sample(cached, min(TOP_K, len(cached)))
+            return recommendation_cache[user_id]
 
         user_profile = self.user_profiles[user_id]
         seen_posts = user_interactions[user_id]
@@ -227,21 +185,14 @@ class Recommender:
             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
+                    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 * 2]
-        selected = random.sample(top_posts, min(TOP_K, len(top_posts)))
-
-        results = []
-        for post_id, score in selected:
-            post_record = post_metadata[post_id].copy()
-            post_record['score'] = score
-            post_record['created_at'] = post_record['created_at'].isoformat()
-            results.append(post_record)
+        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:
@@ -259,10 +210,8 @@ async def background_update():
 def sync_background_update():
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
-    try:
-        loop.run_until_complete(background_update())
-    finally:
-        loop.close()
+    loop.run_until_complete(background_update())
+    loop.close()
 
 @app.get("/recommend/feed")
 async def get_recommendations_handler(user_id: str = Query(...)):