import redis from typing import List, Tuple REDIS_HOST = "localhost" REDIS_PORT = 6379 REDIS_DB = 1 REDIS_DECODE_RESPONSES = True class RedisTagManager: def __init__(self, host="localhost", port=6379, db=0): self.redis_client = redis.Redis( host=host, port=port, db=db, decode_responses=REDIS_DECODE_RESPONSES ) def get_all_tags(self): tags = {} cursor = 0 while True: cursor, keys = self.redis_client.scan(cursor=cursor, match="*", count=1000) for key in keys: try: votes = int(self.redis_client.get(key) or 0) tags[key.decode() if isinstance(key, bytes) else key] = votes except (ValueError, TypeError): continue if cursor == 0: break return tags def get_top_tags(self, n: int) -> List[Tuple[str, int]]: """Get the top N tags sorted by vote count.""" tags = self.get_all_tags() # Sort by vote count descending sorted_tags = sorted(tags.items(), key=lambda x: x[1], reverse=True) return sorted_tags[:n] def get_and_reset_tags(self, n: int) -> List[str]: """Get top N tags and clear the entire database.""" top_tags = self.get_top_tags(n) tag_names = [tag[0] for tag in top_tags] self.redis_client.flushdb() return tag_names