implement history manager

main.py

@@ -75,14 +75,16 @@ async def yalla_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N
     await select_user(context, update.message.chat_id)
 
 
-@bot.command(name="clear", desc="مسح جميع الطلبات")
+@bot.command(name="clear", desc="مسح جميع الطلبات", hidden=True)
 async def clear_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
     if not await util.is_admin(update, context):
         await update.message.reply_text("هذه الخاصية متاحة فقط للأدمن")
         return
 
     global orders
     orders = {}
+    if context.args == "+user_selection_history":
+        userSelector.clear_history()
     await update.message.reply_text("تم مسح جميع الطلبات بنجاح")
 
 

requirements.txt

@@ -30,7 +30,7 @@ pytz==2024.1
 requests==2.31.0
 six==1.16.0
 sniffio==1.3.1
-telebot @ git+https://github.com/mhewedy/telebot@02ac2fa01c87d9ccd8aeb5293e6ee2d0bf299e44
+telebot @ git+https://github.com/mhewedy/telebot@7baf8e4204457f62364bde0a385aed7e6cf4612c
 telegram==0.0.1
 tornado==6.4
 tqdm==4.66.4

test_util.py

@@ -34,7 +34,7 @@ def test_select_single_user(self):
     def test_select_non_excluded_from_2_users(self):
         selector = UserSelector()
         users = ["Alice", "Bob", "Bob", "Bob", "Bob"]
-        selector.history = ["Alice", "Bob", "Alice"]
+        selector.history_manager.history = ["Alice", "Bob", "Alice"]
 
         expected_selections = ["Bob", "Alice", "Bob", "Alice"]
         for expected_selection in expected_selections:
@@ -44,7 +44,7 @@ def test_select_non_excluded_from_2_users(self):
     def test_select_non_excluded_3_users(self):
         selector = UserSelector()
         users = ["Alice", "Alice", "Bob", "Bob", "Charlie", "Charlie", "Charlie"]
-        selector.history = ["Alice", "Bob", "Charlie"]
+        selector.history_manager.history = ["Alice", "Bob", "Charlie"]
 
         expected_selections = ["Alice", "Bob", "Charlie", "Alice", "Bob", "Charlie"]
         for expected_selection in expected_selections:
@@ -54,7 +54,7 @@ def test_select_non_excluded_3_users(self):
     def test_select_non_excluded_3_users_with_ids(self):
         selector = UserSelector()
         users = [(1, "Alice"), (1, "Alice"), (2, "Bob"), (2, "Bob"), (3, "Charlie"), (3, "Charlie"), (3, "Charlie")]
-        selector.history = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
+        selector.history_manager.history = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
 
         expected_selections = [(1, "Alice"), (2, "Bob"), (3, "Charlie"), (1, "Alice"), (2, "Bob"), (3, "Charlie")]
         for expected_selection in expected_selections:
@@ -66,7 +66,7 @@ def test_select_never_picks_last_n_from_history(self):
             selector = UserSelector()
             users = \
                 ["Alice", "Alice", "Bob", "Bob", "Charlie", "Charlie", "Charlie", "Diana", "Diana", "David", "David"]
-            selector.history = ["Alice", "Bob", "Charlie", "Diana"]
+            selector.history_manager.history = ["Alice", "Bob", "Charlie", "Diana"]
             selected = selector.select(users)
             self.assertIn(selected, ["Alice", "Bob", "David"])
 
@@ -77,7 +77,7 @@ def test_history_maintenance(self):
         for _ in range(100):
             selected = selector.select(users)
             selections.append(selected)
-        self.assertEqual(selector.history, selections[-selector.selection_gap:])
+        self.assertEqual(selector.history_manager.history, selections[-selector.selection_gap:])
 
 
 if __name__ == "__main__":

util.py

@@ -1,4 +1,6 @@
+import json
 import logging
+import os
 import random
 import time
 from typing import Any
@@ -35,30 +37,75 @@ async def is_admin(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
     return False
 
 
-class UserSelector:
+class HistoryManager:
     def __init__(self):
         self.history = []
+
+    def load_history(self):
+        pass
+
+    def save_history(self):
+        pass
+
+
+class InMemoryHistoryManager(HistoryManager):
+    def __init__(self):
+        super().__init__()
+
+
+class FileSystemHistoryManager(HistoryManager):
+    def __init__(self, file_path=None):
+        super().__init__()
+        if file_path is None:
+            file_path = os.path.expanduser('~/.lunchy/history.json')
+        self.file_path = file_path
+        self.load_history()
+
+    def load_history(self):
+        directory = os.path.dirname(self.file_path)
+        if directory and not os.path.exists(directory):
+            os.makedirs(directory)
+
+        if os.path.exists(self.file_path):
+            with open(self.file_path, 'r') as file:
+                self.history = json.load(file)
+        else:
+            self.history = []
+            self.save_history()
+
+    def save_history(self):
+        with open(self.file_path, 'w') as file:
+            json.dump(self.history, file)
+
+
+class UserSelector:
+    def __init__(self, history_manager=InMemoryHistoryManager()):
         self.selection_gap = 2
+        self.history_manager = history_manager
 
     def select(self, users):
         if not users:
             raise ValueError('users list should not be empty')
 
         uniq_len = len(list(set(users)))
-        excluded_users = self.history[
-                         len(self.history) - (uniq_len - 1 if uniq_len <= self.selection_gap else self.selection_gap):
+        excluded_users = self.history_manager.history[
+                         len(self.history_manager.history) -
+                         (uniq_len - 1 if uniq_len <= self.selection_gap else self.selection_gap):
                          ]
-
         selected = random.choice(users)
 
         if selected in excluded_users:
             logging.warning(f'{selected} was in excluded history: {excluded_users}, re-selecting...')
             return self.select(users)
         else:
             logging.info(f'users: {users}, selected user is: {selected}')
-            self.history = (self.history + [selected])[-2:]
+            self.history_manager.history = (self.history_manager.history + [selected])[-2:]
+            self.history_manager.save_history()
             return selected
 
+    def clear_history(self):
+        self.history_manager.history = []
+
 
 def get_congrats_msg():
     return random.choice([