import traceback
import time
import redis
# import psycopg2
import pymysql
# from psycopg2 import pool as pgpool
from config import set_config
from log import Log
config_ = set_config()
log_ = Log()
conn_redis = None
conn_filter_redis = None
class RedisHelper(object):
def __init__(self, params=None, redis_info=config_.REDIS_INFO):
"""
初始化redis连接信息
redis_info: redis连接信息, 格式：dict, {'host': '', 'port': '', 'password': ''}
"""
self.redis_info = redis_info
self.host = redis_info['host']
self.port = redis_info['port']
self.password = redis_info['password']
self.params = params
def connect(self):
"""
连接redis
:return: conn
"""
if self.redis_info == config_.REDIS_INFO_FILTER:
global conn_filter_redis
if conn_filter_redis is None:
pool = redis.ConnectionPool(host=self.host,
port=self.port,
password=self.password,
decode_responses=True)
conn = redis.Redis(connection_pool=pool)
conn_filter_redis = conn
return conn_filter_redis
else:
global conn_redis
if conn_redis is None:
pool = redis.ConnectionPool(host=self.host,
port=self.port,
password=self.password,
decode_responses=True)
conn = redis.Redis(connection_pool=pool)
conn_redis = conn
return conn_redis
def key_exists(self, key_name):
"""
判断key是否存在
:param key_name: key
:return: 存在-True, 不存在-False
"""
# start_time = time.time()
conn = self.connect()
res = conn.exists(key_name)
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'get_data_from_redis',
# 'executeTime': (time.time() - start_time) * 1000
# })
return res
def del_keys(self, key_name):
"""
删除key
:param key_name: key
:return: None
"""
conn = self.connect()
conn.delete(key_name)
def get_data_from_redis(self, key_name):
"""
读取redis中的数据
:param key_name: key
:return: data
"""
# start_time = time.time()
conn = self.connect()
if not conn.exists(key_name):
# key不存在
return None
data = conn.get(key_name)
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'get_data_from_redis',
# 'executeTime': (time.time() - start_time) * 1000
# })
return data
def set_data_to_redis(self, key_name, value, expire_time=24*3600):
"""
新增数据
:param key_name: key
:param value: 元素的值 videoId
:param expire_time: 过期时间，单位：s，默认1天
:return: None
"""
# start_time = time.time()
conn = self.connect()
conn.set(key_name, value, ex=int(expire_time))
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'set_data_to_redis',
# 'executeTime': (time.time() - start_time) * 1000})
def add_data_with_zset(self, key_name, data, expire_time=7*24*3600):
"""
新增数据，有序set
:param key_name: key
:param data: 元素的值及对应分数 type-dict {value: score}
:param expire_time: 过期时间，单位：s，默认7天，type-int
:return: None
"""
# start_time = time.time()
conn = self.connect()
conn.zadd(key_name, data)
# 设置过期时间
conn.expire(key_name, int(expire_time))
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'add_data_with_zset',
# 'executeTime': (time.time() - start_time) * 1000
# })
def get_data_zset_with_index(self, key_name, start, end, desc=True, with_scores=False):
"""
根据索引位置获取元素的值
:param key_name: key
:param start: 索引起始点 闭区间，包含start
:param end: 索引结束点 闭区间，包含end
:param desc: 分数排序方式，默认从大到小
:param with_scores: 是否获取元素的分数，默认 False，只获取元素的值
:return: data 元素值列表（不包含分数）,value(videoId)类型转换为int, 包含分数时不进行类型转换
"""
# start_time = time.time()
conn = self.connect()
if not conn.exists(key_name):
return None
data = conn.zrange(key_name, start, end, desc, with_scores)
if with_scores:
data = data
else:
data = [eval(value) for value in data]
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'get_data_zset_with_index',
# 'executeTime': (time.time() - start_time) * 1000
# })
return data
def get_all_data_from_zset(self, key_name, desc=True, with_scores=False):
"""
获取zset中所有元素的值
:param key_name: key
:param desc: 分数排序方式，默认从大到小
:param with_scores: 是否获取元素的分数，默认 False，只获取元素的值
:return: data 元素值列表（不包含分数）,value(videoId)类型转换为int, 包含分数时不进行类型转换
"""
conn = self.connect()
if not conn.exists(key_name):
return None
data = []
start = 0
step = 100
while True:
end = start + step - 1
temp = conn.zrange(key_name, start, end, desc, with_scores)
if not temp:
break
data.extend(temp)
start += step
return data
def get_score_with_value(self, key_name, value):
"""
在zset中，根据元素的value获取对应的score
:param key_name: key
:param value: 元素的值
:return: score value对应的score
"""
conn = self.connect()
if not conn.exists(key_name):
return None
return conn.zscore(key_name, value)
def get_rank_with_value(self, key_name, value, desc=False):
"""
在zset中，根据元素的value获取对应排名
:param key_name: key
:param value: 元素的值
:param desc: 是否倒序 type-bool 默认：False-按照score从小到大
:return: rank value对应的rank，从0开始，不存在返回None
"""
conn = self.connect()
if not conn.exists(key_name):
return None
if desc is True:
return conn.zrevrank(key_name, value)
else:
return conn.zrank(key_name, value)
def update_score_with_value(self, key_name, value, score, expire_time=24*3600):
"""
在zset中，修改元素value对应的score
:param key_name: key
:param value: 元素的值
:param score: value对应的score更新值
:param expire_time: 过期时间，单位：s，默认1天，type-int
"""
conn = self.connect()
if conn.exists(key_name):
conn.zadd(key_name, {value: score})
else:
# key不存在时，需设置过期时间
conn.zadd(key_name, {value: score})
conn.expire(key_name, int(expire_time))
def remove_value_from_zset(self, key_name, value):
"""
删除zset中的指定元素
:param key_name: key
:param value: 元素的值
:return: None
"""
# start_time = time.time()
conn = self.connect()
res = conn.zrem(key_name, value)
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'remove_value_from_zset',
# 'executeTime': (time.time() - start_time) * 1000
# })
return res
def get_index_with_data(self, key_name, value):
"""
根据元素的值获取在有序set中的位置，按照分数倒序（从大到小）
:param key_name: key
:param value: 元素的值
:return: idx 位置索引
"""
# start_time = time.time()
conn = self.connect()
res = conn.zrevrank(key_name, value)
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'get_index_with_data',
# 'executeTime': (time.time() - start_time) * 1000
# })
return res
def get_data_from_set(self, key_name):
"""
获取set中的所有数据
:param key_name: key
:return: data
"""
# start_time = time.time()
conn = self.connect()
if not conn.exists(key_name):
# key不存在
return None
data = []
cursor = 0
while True:
cur, temp = conn.sscan(key_name, cursor=cursor, count=2000)
data.extend(temp)
if cur == 0:
break
cursor = cur
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'get_data_from_set',
# 'executeTime': (time.time() - start_time) * 1000
# })
return list(set(data))
def add_data_with_set(self, key_name, values, expire_time=30*60):
"""
新增数据，set
:param key_name: key
:param values: 要添加的元素 类型-tuple
:param expire_time: 过期时间，单位：s，默认0.5小时 type-int
:return: None
"""
# start_time = time.time()
conn = self.connect()
conn.sadd(key_name, *values)
# 设置过期时间
conn.expire(key_name, int(expire_time))
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'add_data_with_set',
# 'executeTime': (time.time() - start_time) * 1000
# })
def data_exists_with_set(self, key_name, value):
"""
判断元素value是否在集合key_name中
:param key_name: key
:param value: 需判断的元素
:return: 存在-True, 不存在-False
"""
# start_time = time.time()
conn = self.connect()
res = conn.sismember(key_name, value)
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'data_exists_with_set',
# 'executeTime': (time.time() - start_time) * 1000
# })
return res
def get_data_with_count_from_set(self, key_name, count=1):
"""
从set中随机获取元素，并放回
:param key_name: key
:param count: 获取个数, 默认为1
:return:
"""
conn = self.connect()
data = conn.srandmember(name=key_name, number=count)
return data
def remove_value_from_set(self, key_name, values):
"""
删除set中的指定元素
:param key_name: key
:param values: 元素的值, 类型-tuple
:return: None
"""
# start_time = time.time()
conn = self.connect()
conn.srem(key_name, *values)
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'remove_value_from_set',
# 'executeTime': (time.time() - start_time) * 1000
# })
def decr_key(self, key_name, amount=1, expire_time=30*60):
"""
redis自减
:param key_name: key
:param amount: 自减数，默认为1，type-int
:param expire_time: 过期时间，单位：s，默认0.5小时 type-int
:return: None
"""
# start_time = time.time()
conn = self.connect()
conn.decr(name=key_name, amount=amount)
conn.expire(key_name, int(expire_time))
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'decr_key',
# 'executeTime': (time.time() - start_time) * 1000
# })
def incr_key(self, key_name, amount=1, expire_time=30*60):
"""
redis自增
:param key_name: key
:param amount: 自增数，默认为1，type-int
:param expire_time: 过期时间，单位：s，默认0.5小时 type-int
:return: None
"""
# start_time = time.time()
conn = self.connect()
conn.incr(name=key_name, amount=amount)
conn.expire(key_name, int(expire_time))
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'incr_key',
# 'executeTime': (time.time() - start_time) * 1000
# })
def setnx_key(self, key_name, value, expire_time=5*60):
"""
当key不存在时，将value塞入key中，key存在时不做操作
:param key_name: key
:param value: value
:return: 过期时间，单位：s，默认5分钟 type-int
"""
# start_time = time.time()
conn = self.connect()
conn.setnx(name=key_name, value=value)
conn.expire(name=key_name, time=int(expire_time))
# if self.params is not None:
# log_.info({
# 'logTimestamp': int(time.time() * 1000),
# 'request_id': self.params.request_id,
# 'operation': 'setnx_key',
# 'executeTime': (time.time() - start_time) * 1000
# })
def get_batch_key(self, name_list):
conn = self.connect()
res = conn.mget(name_list)
return res
def mget(self, keys):
st_time = time.time()
conn = self.connect()
data = conn.mget(keys=keys)
#print(f"mget time: {(time.time() - st_time) * 1000}")
return data
#hologres_info = config_.HOLOGRES_INFO
#conn = psycopg2.connect(**hologres_info)
#cur = conn.cursor()
# holo连接
# connectPool = pgpool.SimpleConnectionPool(1, 30, **config_.HOLOGRES_INFO)
# class HologresHelper(object):
# def __init__(self):
# """初始化hologres连接信息"""
# #self.hologres_info = config_.HOLOGRES_INFO
#
# def get_data(self, sql):
# #global conn
# # 连接Hologres
# #conn = psycopg2.connect(**self.hologres_info)
# conn = connectPool.getconn()
# # 创建游标
# cur = conn.cursor()
# try:
# # 查询数据
# cur.execute(sql)
# data = cur.fetchall()
# cur.close()
# connectPool.putconn(conn, close=False)
# # 提交事务
# #conn.commit()
# # 释放资源
# #cur.close()
# #conn.close()
# except Exception as e:
# log_.error('hologress error...')
# log_.error(traceback.format_exc())
# cur.close()
# connectPool.putconn(conn, close=False)
# return []
# return data
class MysqlHelper(object):
def __init__(self):
"""
初始化mysql连接信息
"""
self.mysql_info = config_.MYSQL_INFO
def get_data(self, sql):
"""
查询数据
:param sql: sql语句
:return: data
"""
# 连接数据库
conn = pymysql.connect(**self.mysql_info)
# 创建游标
cursor = conn.cursor()
try:
# 执行SQL语句
cursor.execute(sql)
# 获取查询的所有记录
data = cursor.fetchall()
except Exception as e:
return None
# 关闭游标对象
cursor.close()
# 关闭数据库连接
conn.close()
return data
if __name__ == '__main__':
redis_helper = RedisHelper()
# key = 'flow.video.12345.123#112'
# redis_helper.decr_key(key_name=key)
# res = redis_helper.get_data_from_redis(key_name=key)
# print(int(res), type(int(res)))
# data = redis_helper.get_data_zset_with_index(key_name=config_.BOTTOM_KEY_NAME, start=0, end=-1)
# print(len(data))
# key_name = 'com.weiqu.video.hot.recommend.previewed.4.weixin_openid_otjoB5VG780SB4aVjYqBBNLb - X6M'
# values = (6134455, 9772930, 9912678, 9901969, 9926876, 9904203, 2384831, 9932272, 9737653, 9925240)
# key_name = 'com.weiqu.video.hot.recommend.previewed.4.abcd1'
# values = (9902612, 9905573, 9928264, 9932148, 9809440, 9919900, 6093379, 9917093, 9793537, 9814345)
# redis_helper.add_data_with_set(key_name=key_name, values=values, expire_time=30 * 60)
# res = redis_helper.get_data_from_set(key_name=key_name)
res = redis_helper.remove_value_from_zset(
key_name="recall:item:score:region:dup3:24h:110000:data1:rule4:20230315:14",
value=111111)
print(res)