Spaces:
Sleeping
Sleeping
File size: 3,588 Bytes
3b86501 8ebe686 3b86501 8ebe686 3b86501 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
#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")
|