dm/ai-slibra-assistant/database/redis.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time    : 2024/10/17 15:46
@Author  : cao
@File    : redis.py.py
@Desc    : 
"""

from config import get_logger
import redis
import os
from typing import (Text,
                    )
import json

logger = get_logger()


def get_redis_config(env='prod'):
    if env == "prod":
        redis_host = os.getenv("RADIS_HOST", "192.168.100.159")
        redis_password = os.getenv("redis_password", "^YHN&UJM")
        redis_port = 6379
        redis_db = 0
    else:
        redis_host = os.getenv("RADIS_HOST", "10.0.0.24")
        redis_password = os.getenv("redis_password", "^YHN&UJM")
        redis_port = 6379
        redis_db = 0
    return redis_host, redis_password, redis_port, redis_db

redis_host, redis_password, redis_port, redis_db = get_redis_config()

pool = redis.ConnectionPool(host=redis_host,
                            port=redis_port,
                            password=redis_password,
                            db=redis_db,
                            socket_timeout=1000,
                            decode_responses=True
                            )

r = redis.Redis(connection_pool=pool)

def get_status_cache(uid: Text, bot_id: Text, label: Text, session_id: Text):
    # 获取用户当前状态
    from scene import Dialog, RobotService
    scene = None
    try:
        name = f"{bot_id}_{label}"
        key = f"{uid}_{session_id}"
        records = r.hget(name, key)
        if records:
            logger.info(f"获取缓存信息成功:uid:{uid},session:{session_id}")
            res = json.loads(records)
            logger.info(f"load cache: {res}")
            case = RobotService(**res)
            scene = Dialog(case)
            return scene
        else:
            logger.info(f"没有缓存信息:uid:{uid},session:{session_id}")
            return
    except Exception as e:
        logger.error("获取缓存信息失败%s", e)
        return



def del_cache(uid: Text, bot_id: Text, label: Text, session_id: Text):
    try:
        name = f"{bot_id}_{label}"
        key = f"{uid}_{session_id}"
        r.hdel(name, key)
        logger.info("删除缓存状态信息success")
    except Exception as e:
        logger.error("删除缓存状态信息%s", e)

def with_status_cache(uid: Text, bot_id: Text, label: Text, session_id: Text, scene):
    try:
        name = f"{bot_id}_{label}"
        key = f"{uid}_{session_id}"
        value = json.dumps(scene.case, ensure_ascii=False, default=lambda obj: obj.__dict__)
        logger.info(f"save cache：uid:{uid};sessionid:{session_id}:code:{scene.case.code}")
        r.hset(name, key, value)
        r.expire(name, 3600)
    except Exception as e:
        logger.error("redis缓存状态%s", e)


def get_tracking(uid: Text):
    try:
        name = f"{uid}_tracking"
        key = f"{uid}"
        content = r.hget(name, key)
        if content:
            return content
    except Exception as e:
        logger.error(e)
        return None


if __name__ == "__main__":
    set_cache("yxxxx2", "10001","status", "yxx", "2.10")