# 方法1：通過(guò)django-redis獲取連接
from django_redis import get_redis_connection

# 獲取默認(rèn)緩存對(duì)應(yīng)的Redis連接
redis_conn = get_redis_connection("default")

# 獲取特定緩存的連接
session_conn = get_redis_connection("session")

# 方法2：直接創(chuàng)建連接（不推薦，缺少連接池管理）
import redis
redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    password=None
)

def string_operations():
    """Redis字符串操作"""
    redis_conn = get_redis_connection("default")
    
    # 1. 設(shè)置帶過(guò)期時(shí)間的鍵
    redis_conn.setex("session:user:123", 3600, "session_data")  # 1小時(shí)后過(guò)期
    
    # 2. 設(shè)置多個(gè)鍵
    redis_conn.mset({"key1": "value1", "key2": "value2"})
    
    # 3. 獲取字符串的一部分
    redis_conn.set("message", "Hello World")
    part = redis_conn.getrange("message", 0, 4)  # "Hello"
    
    # 4. 追加字符串
    redis_conn.append("message", " Redis!")
    
    # 5. 獲取字符串長(zhǎng)度
    length = redis_conn.strlen("message")
    
    # 6. 設(shè)置鍵的值并返回舊值
    old_value = redis_conn.getset("counter", "100")
    
    # 7. 位操作
    redis_conn.setbit("user:online:2023-10-01", 123, 1)  # 用戶123在線
    redis_conn.getbit("user:online:2023-10-01", 123)  # 檢查是否在線
    
    return {
        "part": part,
        "length": length,
        "old_value": old_value
    }

def list_operations():
    """Redis列表操作"""
    redis_conn = get_redis_connection("default")
    
    # 1. 左側(cè)推入元素
    redis_conn.lpush("recent_users", "user_123", "user_456", "user_789")
    
    # 2. 右側(cè)推入元素
    redis_conn.rpush("task_queue", "task1", "task2", "task3")
    
    # 3. 左側(cè)彈出元素
    first_user = redis_conn.lpop("recent_users")
    
    # 4. 右側(cè)彈出元素
    last_task = redis_conn.rpop("task_queue")
    
    # 5. 獲取列表長(zhǎng)度
    length = redis_conn.llen("recent_users")
    
    # 6. 獲取指定范圍的元素
    users = redis_conn.lrange("recent_users", 0, 9)  # 前10個(gè)元素
    
    # 7. 移除元素
    redis_conn.lrem("recent_users", 1, "user_123")  # 刪除1個(gè)user_123
    
    # 8. 通過(guò)索引獲取元素
    user = redis_conn.lindex("recent_users", 0)  # 獲取第一個(gè)
    
    # 9. 設(shè)置指定索引的值
    redis_conn.lset("recent_users", 0, "new_user")
    
    return {
        "first_user": first_user,
        "last_task": last_task,
        "length": length,
        "users": users
    }

class MessageQueue:
    """基于Redis列表的消息隊(duì)列"""
    
    def __init__(self, queue_name="default_queue"):
        self.redis_conn = get_redis_connection("default")
        self.queue_name = queue_name
    
    def enqueue(self, message):
        """入隊(duì)"""
        import json
        message_str = json.dumps(message)
        return self.redis_conn.rpush(self.queue_name, message_str)
    
    def dequeue(self, timeout=0):
        """出隊(duì)（阻塞）"""
        import json
        result = self.redis_conn.blpop(self.queue_name, timeout=timeout)
        if result:
            queue_name, message_str = result
            return json.loads(message_str)
        return None
    
    def size(self):
        """隊(duì)列大小"""
        return self.redis_conn.llen(self.queue_name)
    
    def peek(self):
        """查看但不移除"""
        import json
        message_str = self.redis_conn.lindex(self.queue_name, 0)
        if message_str:
            return json.loads(message_str)
        return None

# 使用示例
def process_messages():
    """處理消息隊(duì)列"""
    queue = MessageQueue("email_queue")
    
    # 生產(chǎn)者
    queue.enqueue({
        "to": "user@example.com",
        "subject": "Welcome",
        "body": "Hello!"
    })
    
    # 消費(fèi)者
    while True:
        message = queue.dequeue(timeout=5)  # 阻塞5秒
        if message:
            # 處理消息
            send_email(**message)
        else:
            # 沒(méi)有消息，休息一下
            import time
            time.sleep(1)

def set_operations():
    """Redis集合操作"""
    redis_conn = get_redis_connection("default")
    
    # 1. 添加元素
    redis_conn.sadd("user_tags:123", "python", "django", "redis")
    redis_conn.sadd("user_tags:456", "python", "javascript", "vue")
    
    # 2. 獲取所有元素
    tags_123 = redis_conn.smembers("user_tags:123")
    
    # 3. 判斷元素是否存在
    is_member = redis_conn.sismember("user_tags:123", "python")
    
    # 4. 移除元素
    redis_conn.srem("user_tags:123", "redis")
    
    # 5. 獲取集合大小
    size = redis_conn.scard("user_tags:123")
    
    # 6. 隨機(jī)獲取元素
    random_tag = redis_conn.srandmember("user_tags:123", 1)
    
    # 7. 集合運(yùn)算
    # 交集：都喜歡的標(biāo)簽
    common_tags = redis_conn.sinter("user_tags:123", "user_tags:456")
    
    # 并集：所有標(biāo)簽
    all_tags = redis_conn.sunion("user_tags:123", "user_tags:456")
    
    # 差集：用戶123有但456沒(méi)有的標(biāo)簽
    diff_tags = redis_conn.sdiff("user_tags:123", "user_tags:456")
    
    return {
        "tags_123": tags_123,
        "is_member": is_member,
        "size": size,
        "common_tags": common_tags
    }

class UserTagSystem:
    """基于Redis集合的用戶標(biāo)簽系統(tǒng)"""
    
    def __init__(self):
        self.redis_conn = get_redis_connection("default")
    
    def add_tag(self, user_id, tag):
        """為用戶添加標(biāo)簽"""
        return self.redis_conn.sadd(f"user_tags:{user_id}", tag)
    
    def remove_tag(self, user_id, tag):
        """移除用戶標(biāo)簽"""
        return self.redis_conn.srem(f"user_tags:{user_id}", tag)
    
    def get_user_tags(self, user_id):
        """獲取用戶所有標(biāo)簽"""
        return self.redis_conn.smembers(f"user_tags:{user_id}")
    
    def has_tag(self, user_id, tag):
        """檢查用戶是否有某個(gè)標(biāo)簽"""
        return self.redis_conn.sismember(f"user_tags:{user_id}", tag)
    
    def find_users_with_tag(self, tag):
        """查找具有某個(gè)標(biāo)簽的所有用戶"""
        # 需要遍歷所有用戶，實(shí)際應(yīng)用中可以使用反向索引
        pattern = "user_tags:*"
        matching_users = []
        
        # 使用SCAN避免阻塞
        cursor = 0
        while True:
            cursor, keys = self.redis_conn.scan(cursor, pattern, count=100)
            for key in keys:
                if self.redis_conn.sismember(key, tag):
                    user_id = key.decode().split(":")[1]
                    matching_users.append(user_id)
            
            if cursor == 0:
                break
        
        return matching_users
    
    def get_common_tags(self, user_ids):
        """獲取多個(gè)用戶的共同標(biāo)簽"""
        if len(user_ids) < 2:
            return []
        
        keys = [f"user_tags:{user_id}" for user_id in user_ids]
        return self.redis_conn.sinter(*keys)
    
    def get_recommended_tags(self, user_id):
        """根據(jù)相似用戶推薦標(biāo)簽"""
        user_tags = self.get_user_tags(user_id)
        if not user_tags:
            return []
        
        # 查找有相同標(biāo)簽的用戶
        similar_users = []
        for tag in user_tags:
            users_with_tag = self.find_users_with_tag(tag)
            similar_users.extend(users_with_tag)
        
        # 去重并排除自己
        similar_users = set(similar_users) - {user_id}
        
        # 收集相似用戶的標(biāo)簽
        all_tags = set()
        for similar_user in list(similar_users)[:10]:  # 只看前10個(gè)相似用戶
            tags = self.get_user_tags(similar_user)
            all_tags.update(tags)
        
        # 排除用戶已有的標(biāo)簽
        recommended = all_tags - set(user_tags)
        
        return list(recommended)[:10]  # 返回前10個(gè)推薦標(biāo)簽

def sorted_set_operations():
    """Redis有序集合操作"""
    redis_conn = get_redis_connection("default")
    
    # 1. 添加元素（帶分?jǐn)?shù)）
    redis_conn.zadd("leaderboard", {
        "player1": 1000,
        "player2": 1500,
        "player3": 800,
        "player4": 2000
    })
    
    # 2. 增加分?jǐn)?shù)
    redis_conn.zincrby("leaderboard", 100, "player1")  # player1增加100分
    
    # 3. 按分?jǐn)?shù)升序獲取
    ascending = redis_conn.zrange("leaderboard", 0, -1, withscores=True)
    
    # 4. 按分?jǐn)?shù)降序獲取
    descending = redis_conn.zrevrange("leaderboard", 0, -1, withscores=True)
    
    # 5. 獲取排名
    rank = redis_conn.zrevrank("leaderboard", "player1")  # 降序排名，0為第一
    
    # 6. 獲取分?jǐn)?shù)
    score = redis_conn.zscore("leaderboard", "player1")
    
    # 7. 獲取分?jǐn)?shù)范圍內(nèi)的元素
    top_players = redis_conn.zrangebyscore("leaderboard", 1000, 3000, withscores=True)
    
    # 8. 移除元素
    redis_conn.zrem("leaderboard", "player4")
    
    # 9. 獲取集合大小
    size = redis_conn.zcard("leaderboard")
    
    # 10. 統(tǒng)計(jì)分?jǐn)?shù)區(qū)間的元素?cái)?shù)量
    count = redis_conn.zcount("leaderboard", 1000, 3000)
    
    return {
        "rank": rank,
        "score": score,
        "top_3": descending[:3],
        "count": count
    }

class Leaderboard:
    """基于Redis有序集合的排行榜系統(tǒng)"""
    
    def __init__(self, leaderboard_name="default_leaderboard"):
        self.redis_conn = get_redis_connection("default")
        self.name = leaderboard_name
    
    def add_score(self, player_id, score):
        """添加或更新分?jǐn)?shù)"""
        return self.redis_conn.zincrby(self.name, score, player_id)
    
    def get_top_n(self, n=10):
        """獲取前N名"""
        return self.redis_conn.zrevrange(
            self.name, 0, n-1, withscores=True
        )
    
    def get_player_rank(self, player_id):
        """獲取玩家排名（從1開(kāi)始）"""
        rank = self.redis_conn.zrevrank(self.name, player_id)
        if rank is not None:
            return rank + 1
        return None
    
    def get_player_score(self, player_id):
        """獲取玩家分?jǐn)?shù)"""
        return self.redis_conn.zscore(self.name, player_id)
    
    def get_players_around(self, player_id, range_size=5):
        """獲取玩家附近的排名"""
        rank = self.get_player_rank(player_id)
        if rank is None:
            return []
        
        start = max(0, rank - 1 - range_size)
        end = rank - 1 + range_size
        
        players = self.redis_conn.zrevrange(
            self.name, start, end, withscores=True
        )
        
        return [
            {"player": player, "score": score, "rank": idx + start + 1}
            for idx, (player, score) in enumerate(players)
        ]
    
    def get_season_leaderboard(self, season_id, top_n=100):
        """獲取賽季排行榜"""
        season_key = f"{self.name}:season:{season_id}"
        
        # 檢查是否有緩存的賽季排行榜
        cached = self.redis_conn.get(season_key)
        if cached:
            import json
            return json.loads(cached)
        
        # 計(jì)算賽季排行榜
        season_end = int(season_id)  # 假設(shè)season_id是時(shí)間戳
        season_start = season_end - 86400 * 30  # 30天
        
        # 獲取賽季期間有分?jǐn)?shù)的玩家
        all_players = self.redis_conn.zrevrangebyscore(
            f"{self.name}:daily:{season_end}",
            season_start, season_end,
            withscores=False
        )
        
        # 計(jì)算每個(gè)玩家的賽季總分
        pipe = self.redis_conn.pipeline()
        for player in all_players:
            pipe.zscore(f"{self.name}:daily", player)
        
        scores = pipe.execute()
        
        # 創(chuàng)建有序集合
        season_data = {player: score for player, score in zip(all_players, scores) if score}
        
        if season_data:
            season_leaderboard_key = f"{self.name}:season_temp:{season_id}"
            self.redis_conn.zadd(season_leaderboard_key, season_data)
            
            # 獲取前N名
            top_players = self.redis_conn.zrevrange(
                season_leaderboard_key, 0, top_n-1, withscores=True
            )
            
            # 刪除臨時(shí)鍵
            self.redis_conn.delete(season_leaderboard_key)
            
            # 緩存結(jié)果
            import json
            self.redis_conn.setex(
                season_key,
                3600,  # 緩存1小時(shí)
                json.dumps(top_players)
            )
            
            return top_players
        
        return []


# 游戲積分系統(tǒng)示例
class GameScoreSystem:
    """游戲積分系統(tǒng)"""
    
    def __init__(self, game_id):
        self.redis_conn = get_redis_connection("default")
        self.game_id = game_id
        self.leaderboard = Leaderboard(f"game:{game_id}:leaderboard")
    
    def record_score(self, user_id, score, level, time_spent):
        """記錄游戲得分"""
        import time
        timestamp = int(time.time())
        
        # 1. 更新總排行榜
        self.leaderboard.add_score(user_id, score)
        
        # 2. 記錄詳細(xì)得分（使用哈希）
        score_key = f"game:{self.game_id}:score:{user_id}:{timestamp}"
        self.redis_conn.hmset(score_key, {
            "score": score,
            "level": level,
            "time_spent": time_spent,
            "timestamp": timestamp
        })
        
        # 設(shè)置24小時(shí)過(guò)期
        self.redis_conn.expire(score_key, 86400)
        
        # 3. 更新每日排行榜
        date_str = time.strftime("%Y%m%d")
        daily_key = f"game:{self.game_id}:daily:{date_str}"
        self.redis_conn.zincrby(daily_key, score, user_id)
        
        # 4. 更新關(guān)卡排行榜
        level_key = f"game:{self.game_id}:level:{level}"
        self.redis_conn.zincrby(level_key, 1, user_id)  # 通關(guān)次數(shù)
        
        return {
            "rank": self.leaderboard.get_player_rank(user_id),
            "score": self.leaderboard.get_player_score(user_id)
        }

def hash_operations():
    """Redis哈希操作"""
    redis_conn = get_redis_connection("default")
    
    # 1. 設(shè)置單個(gè)字段
    redis_conn.hset("user:123", "name", "張三")
    
    # 2. 設(shè)置多個(gè)字段
    redis_conn.hmset("user:123", {
        "age": 25,
        "email": "zhangsan@example.com",
        "city": "北京"
    })
    
    # 3. 獲取單個(gè)字段
    name = redis_conn.hget("user:123", "name")
    
    # 4. 獲取多個(gè)字段
    fields = redis_conn.hmget("user:123", ["name", "age", "email"])
    
    # 5. 獲取所有字段
    all_fields = redis_conn.hgetall("user:123")
    
    # 6. 獲取所有字段名
    field_names = redis_conn.hkeys("user:123")
    
    # 7. 獲取所有字段值
    field_values = redis_conn.hvals("user:123")
    
    # 8. 檢查字段是否存在
    exists = redis_conn.hexists("user:123", "name")
    
    # 9. 刪除字段
    redis_conn.hdel("user:123", "city")
    
    # 10. 獲取字段數(shù)量
    field_count = redis_conn.hlen("user:123")
    
    # 11. 增加數(shù)值字段的值
    redis_conn.hincrby("user:123", "age", 1)  # 增加1歲
    
    return {
        "name": name,
        "fields": fields,
        "exists": exists,
        "field_count": field_count
    }

class ShoppingCart:
    """基于Redis哈希的購(gòu)物車系統(tǒng)"""
    
    def __init__(self, cart_id=None):
        self.redis_conn = get_redis_connection("default")
        self.cart_id = cart_id or f"cart:{int(time.time())}"
    
    def add_item(self, product_id, quantity=1, price=None):
        """添加商品到購(gòu)物車"""
        cart_key = f"cart:items:{self.cart_id}"
        
        # 獲取當(dāng)前數(shù)量
        current_quantity = self.redis_conn.hget(cart_key, product_id)
        if current_quantity:
            quantity = int(current_quantity) + quantity
        
        # 更新購(gòu)物車
        self.redis_conn.hset(cart_key, product_id, quantity)
        
        # 如果提供了價(jià)格，存儲(chǔ)到商品信息
        if price is not None:
            product_key = f"cart:product_info:{self.cart_id}"
            self.redis_conn.hset(product_key, product_id, price)
        
        return quantity
    
    def remove_item(self, product_id, quantity=None):
        """從購(gòu)物車移除商品"""
        cart_key = f"cart:items:{self.cart_id}"
        
        if quantity is None:
            # 移除整個(gè)商品
            self.redis_conn.hdel(cart_key, product_id)
            
            # 同時(shí)移除價(jià)格信息
            product_key = f"cart:product_info:{self.cart_id}"
            self.redis_conn.hdel(product_key, product_id)
        else:
            # 減少數(shù)量
            current_quantity = self.redis_conn.hget(cart_key, product_id)
            if current_quantity:
                new_quantity = int(current_quantity) - quantity
                if new_quantity > 0:
                    self.redis_conn.hset(cart_key, product_id, new_quantity)
                else:
                    self.remove_item(product_id)
    
    def get_items(self):
        """獲取購(gòu)物車所有商品"""
        cart_key = f"cart:items:{self.cart_id}"
        items = self.redis_conn.hgetall(cart_key)
        
        product_key = f"cart:product_info:{self.cart_id}"
        prices = self.redis_conn.hgetall(product_key)
        
        result = []
        for product_id, quantity in items.items():
            product_id = product_id.decode() if isinstance(product_id, bytes) else product_id
            quantity = int(quantity)
            
            price = prices.get(product_id.encode() if isinstance(product_id, str) else product_id)
            if price:
                price = float(price.decode() if isinstance(price, bytes) else price)
            
            result.append({
                "product_id": product_id,
                "quantity": quantity,
                "price": price,
                "subtotal": price * quantity if price else None
            })
        
        return result
    
    def get_total(self):
        """計(jì)算購(gòu)物車總價(jià)"""
        items = self.get_items()
        return sum(item.get("subtotal", 0) for item in items)
    
    def clear(self):
        """清空購(gòu)物車"""
        cart_key = f"cart:items:{self.cart_id}"
        product_key = f"cart:product_info:{self.cart_id}"
        
        self.redis_conn.delete(cart_key, product_key)
    
    def get_item_count(self):
        """獲取購(gòu)物車商品總數(shù)"""
        cart_key = f"cart:items:{self.cart_id}"
        items = self.redis_conn.hvals(cart_key)
        return sum(int(q) for q in items) if items else 0
    
    def checkout(self):
        """結(jié)賬"""
        items = self.get_items()
        
        if not items:
            return {"success": False, "message": "購(gòu)物車為空"}
        
        # 生成訂單
        import uuid
        order_id = str(uuid.uuid4())
        
        # 保存訂單
        order_key = f"order:{order_id}"
        self.redis_conn.hmset(order_key, {
            "cart_id": self.cart_id,
            "total": self.get_total(),
            "created_at": int(time.time()),
            "status": "pending"
        })
        
        # 保存訂單商品
        for item in items:
            self.redis_conn.hset(
                f"{order_key}:items",
                item["product_id"],
                item["quantity"]
            )
        
        # 清空購(gòu)物車
        self.clear()
        
        return {
            "success": True,
            "order_id": order_id,
            "total": self.get_total(),
            "item_count": len(items)
        }

class SocialNetwork:
    """基于Redis的社交關(guān)系系統(tǒng)"""
    
    def __init__(self):
        self.redis_conn = get_redis_connection("default")
    
    def follow(self, follower_id, followee_id):
        """關(guān)注用戶"""
        # 添加到關(guān)注列表（集合）
        self.redis_conn.sadd(f"user:{follower_id}:following", followee_id)
        self.redis_conn.sadd(f"user:{followee_id}:followers", follower_id)
        
        # 記錄時(shí)間線（有序集合）
        import time
        timestamp = time.time()
        
        # 獲取關(guān)注用戶的動(dòng)態(tài)
        posts_key = f"user:{followee_id}:posts"
        posts = self.redis_conn.zrevrangebyscore(
            posts_key, timestamp - 86400 * 7, timestamp,  # 最近7天的帖子
            withscores=True
        )
        
        # 添加到關(guān)注者的時(shí)間線
        timeline_key = f"user:{follower_id}:timeline"
        for post_id, post_time in posts:
            self.redis_conn.zadd(timeline_key, {post_id: post_time})
        
        # 限制時(shí)間線長(zhǎng)度
        self.redis_conn.zremrangebyrank(timeline_key, 0, -1000)  # 只保留最新的1000條
    
    def unfollow(self, follower_id, followee_id):
        """取消關(guān)注"""
        self.redis_conn.srem(f"user:{follower_id}:following", followee_id)
        self.redis_conn.srem(f"user:{followee_id}:followers", follower_id)
    
    def get_followers(self, user_id, page=1, per_page=20):
        """獲取粉絲列表"""
        key = f"user:{user_id}:followers"
        start = (page - 1) * per_page
        end = start + per_page - 1
        
        # 使用集合獲取粉絲
        followers = self.redis_conn.smembers(key)
        
        # 獲取粉絲詳細(xì)信息
        followers_list = []
        for follower_id in list(followers)[start:end+1]:
            user_info = self.redis_conn.hgetall(f"user:{follower_id}")
            followers_list.append(user_info)
        
        return {
            "followers": followers_list,
            "total": self.redis_conn.scard(key),
            "page": page,
            "per_page": per_page
        }
    
    def get_mutual_followers(self, user1_id, user2_id):
        """獲取共同關(guān)注的人"""
        key1 = f"user:{user1_id}:following"
        key2 = f"user:{user2_id}:following"
        
        return self.redis_conn.sinter(key1, key2)
    
    def post_update(self, user_id, content):
        """發(fā)布動(dòng)態(tài)"""
        import time
        import uuid
        
        post_id = str(uuid.uuid4())
        timestamp = time.time()
        
        # 存儲(chǔ)帖子內(nèi)容（哈希）
        post_key = f"post:{post_id}"
        self.redis_conn.hmset(post_key, {
            "id": post_id,
            "user_id": user_id,
            "content": content,
            "timestamp": timestamp,
            "likes": 0,
            "comments": 0
        })
        
        # 添加到用戶的帖子列表（有序集合）
        user_posts_key = f"user:{user_id}:posts"
        self.redis_conn.zadd(user_posts_key, {post_id: timestamp})
        
        # 添加到粉絲的時(shí)間線
        followers_key = f"user:{user_id}:followers"
        followers = self.redis_conn.smembers(followers_key)
        
        for follower_id in followers:
            timeline_key = f"user:{follower_id}:timeline"
            self.redis_conn.zadd(timeline_key, {post_id: timestamp})
            self.redis_conn.zremrangebyrank(timeline_key, 0, -1000)
        
        return post_id
    
    def get_timeline(self, user_id, page=1, per_page=20):
        """獲取時(shí)間線"""
        timeline_key = f"user:{user_id}:timeline"
        start = (page - 1) * per_page
        end = start + per_page - 1
        
        # 獲取帖子ID
        post_ids = self.redis_conn.zrevrange(timeline_key, start, end)
        
        # 獲取帖子詳情
        posts = []
        for post_id in post_ids:
            post = self.redis_conn.hgetall(f"post:{post_id}")
            if post:
                posts.append(post)
        
        return {
            "posts": posts,
            "total": self.redis_conn.zcard(timeline_key),
            "page": page,
            "per_page": per_page
        }

class RealTimeStatistics:
    """實(shí)時(shí)統(tǒng)計(jì)系統(tǒng)"""
    
    def __init__(self):
        self.redis_conn = get_redis_connection("default")
    
    def record_event(self, event_type, user_id=None, data=None):
        """記錄事件"""
        import time
        timestamp = int(time.time())
        
        # 1. 記錄到全局事件流（列表）
        event_id = f"{timestamp}:{user_id}:{event_type}"
        self.redis_conn.lpush("events:stream", event_id)
        
        # 2. 按類型計(jì)數(shù)（哈希）
        today = time.strftime("%Y%m%d")
        daily_key = f"stats:events:{today}"
        self.redis_conn.hincrby(daily_key, event_type, 1)
        
        # 3. 記錄用戶事件（有序集合）
        if user_id:
            user_events_key = f"user:{user_id}:events"
            self.redis_conn.zadd(user_events_key, {event_type: timestamp})
            
            # 用戶最后活躍時(shí)間
            self.redis_conn.hset(f"user:{user_id}:activity", "last_active", timestamp)
        
        # 4. 按小時(shí)統(tǒng)計(jì)（哈希）
        hour = time.strftime("%Y%m%d%H")
        hour_key = f"stats:hourly:{hour}"
        self.redis_conn.hincrby(hour_key, event_type, 1)
        
        return event_id
    
    def get_daily_stats(self, date_str=None):
        """獲取每日統(tǒng)計(jì)"""
        import time
        if not date_str:
            date_str = time.strftime("%Y%m%d")
        
        key = f"stats:events:{date_str}"
        return self.redis_conn.hgetall(key)
    
    def get_hourly_trend(self, event_type, hours=24):
        """獲取小時(shí)趨勢(shì)"""
        import time
        from datetime import datetime, timedelta
        
        now = datetime.now()
        trend = []
        
        for i in range(hours):
            hour = now - timedelta(hours=i)
            hour_str = hour.strftime("%Y%m%d%H")
            key = f"stats:hourly:{hour_str}"
            
            count = self.redis_conn.hget(key, event_type)
            trend.append({
                "hour": hour_str,
                "count": int(count) if count else 0
            })
        
        return list(reversed(trend))  # 從早到晚
    
    def get_active_users(self, minutes=5):
        """獲取活躍用戶（最近N分鐘有活動(dòng)的用戶）"""
        import time
        
        # 使用集合存儲(chǔ)活躍用戶
        active_users_key = "users:active"
        
        # 獲取所有用戶
        pattern = "user:*:activity"
        active_users = set()
        
        cursor = 0
        while True:
            cursor, keys = self.redis_conn.scan(cursor, pattern, count=100)
            
            for key in keys:
                # 解析用戶ID
                user_id = key.decode().split(":")[1]
                
                # 檢查最后活躍時(shí)間
                last_active = self.redis_conn.hget(key, "last_active")
                if last_active:
                    last_active = int(last_active)
                    if time.time() - last_active < minutes * 60:
                        active_users.add(user_id)
            
            if cursor == 0:
                break
        
        return list(active_users)

def use_pipeline():
    """使用管道批量操作"""
    redis_conn = get_redis_connection("default")
    
    # 創(chuàng)建管道
    pipe = redis_conn.pipeline()
    
    # 批量添加命令
    for i in range(100):
        pipe.set(f"key:{i}", f"value:{i}")
        pipe.expire(f"key:{i}", 3600)
    
    # 一次執(zhí)行所有命令
    results = pipe.execute()
    
    return len(results)  # 200個(gè)結(jié)果

def use_lua_script():
    """使用Lua腳本實(shí)現(xiàn)原子操作"""
    redis_conn = get_redis_connection("default")
    
    # 原子增加分?jǐn)?shù)并更新排行榜
    lua_script = """
    local player = KEYS[1]
    local score = tonumber(ARGV[1])
    local leaderboard = KEYS[2]
    
    -- 增加分?jǐn)?shù)
    local new_score = redis.call('ZINCRBY', leaderboard, score, player)
    
    -- 獲取排名
    local rank = redis.call('ZREVRANK', leaderboard, player)
    
    -- 返回結(jié)果
    return {new_score, rank}
    """
    
    # 執(zhí)行腳本
    result = redis_conn.eval(
        lua_script,
        2,  # 2個(gè)KEYS
        "player_123",  # KEYS[1]
        "leaderboard",  # KEYS[2]
        100  # ARGV[1]
    )
    
    return {
        "new_score": float(result[0]),
        "rank": int(result[1]) + 1 if result[1] is not None else None
    }