Spaces:
Sleeping
Sleeping
| #import redis | |
| from modules.utils import compare_date_with_today | |
| import pandas as pd | |
| import random | |
| import uuid | |
| #load_dotenv('../../.env') | |
| #redis_pw = os.getenv("REDIS_PASSWORD") | |
| #try: | |
| # r = redis.Redis( | |
| # host='redis-12097.c328.europe-west3-1.gce.cloud.redislabs.com', | |
| # port=12097, | |
| # password=redis_pw) | |
| #except: | |
| # print("Redis not available - working with memory alone") | |
| # r = None | |
| #on_memory_user_bank = {} | |
| #on_memory_boost_bank = load_csv_to_dict('../data/boost_bank.csv') | |
| class BoostDatabase: | |
| def __init__(self, csv_path): | |
| """Initialize the database by loading boosts from a CSV file.""" | |
| self.boosts_bank = {} | |
| boosts = pd.read_csv(csv_path) | |
| for target_topic in boosts['target_topic'].unique(): | |
| subset = boosts.query('target_topic == @target_topic') | |
| self.boosts_bank[target_topic] = { | |
| 'twitter': [{'id': str(uuid.uuid4()), 'url': url} for url in subset['url_x'].dropna()], | |
| 'facebook': [{'id': str(uuid.uuid4()), 'url': url} for url in subset['url_facebook'].dropna()], | |
| 'reddit': [{'id': str(uuid.uuid4()), 'url': url} for url in subset['url_reddit_new'].dropna()], | |
| } | |
| def get_random_boost(self, topic, platform, blacklist_ids): | |
| """ | |
| Retrieve a random boost item (URL and UUID) for a given topic and platform, | |
| excluding any items with UUIDs in the blacklist. | |
| """ | |
| if topic in self.boosts_bank and platform in self.boosts_bank[topic]: | |
| # Filter out any items that have an ID in the blacklist | |
| valid_items = [item for item in self.boosts_bank[topic][platform] if item['id'] not in blacklist_ids] | |
| if valid_items: | |
| return random.choice(valid_items) | |
| else: | |
| return "No available boosts that haven't been seen." | |
| else: | |
| return "Topic or platform not found." | |
| class User: | |
| def __init__(self, user_id: str, last_boost: str, boosts=None): | |
| if boosts is None: | |
| boosts = [] | |
| self.user_id = user_id | |
| self.boosts = boosts | |
| self.last_boost = last_boost | |
| def add_boost(self, boost): | |
| """Append a new boost to the list of boosts.""" | |
| self.boosts.append(boost) | |
| def is_boosted_today(self): | |
| return compare_date_with_today(self.last_boost) | |
| def update_boosted_today(self, status): | |
| """Update the boosted_today status.""" | |
| self.boosted_today = status | |
| #def __repr__(self): | |
| # return f"User(boosts={self.boosts}, boosted_today={self.boosted_today})" | |
| class UserDatabase: | |
| def __init__(self): | |
| self.users = {} # Stores user_id mapped to User objects | |
| def add_user(self, user_id, user): | |
| """Add a new user to the database.""" | |
| if user_id not in self.users: | |
| self.users[user_id] = user | |
| else: | |
| print("User already exists with this ID") | |
| def get_user(self, user_id): | |
| """Retrieve a user by user_id.""" | |
| return self.users.get(user_id, None) | |
| def add_boost_to_user(self, user_id, boost): | |
| """Append a boost to a user's list of boosts.""" | |
| user = self.get_user(user_id) | |
| if user: | |
| user.add_boost(boost) | |
| else: | |
| print("User not found") | |
| def update_user_boosted_today(self, user_id, date): | |
| """Update the boosted_today status for a user.""" | |
| user = self.get_user(user_id) | |
| if user: | |
| user.update_boosted_today(date) | |
| else: | |
| print("User not found") | |