#!/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")