Gecko [ www.jdcpatala.org ]

Name	Size	Permission
__init__.py	1 B	-rw-r--r--
base.py	21.68 KB	-rw-r--r--
mongodbModel.py	38.86 KB	-rw-r--r--
pgsqlModel.py	31.87 KB	-rw-r--r--
redisModel.py	20.5 KB	-rw-r--r--
sqliteModel.py	667 B	-rw-r--r--
sqlserverModel.py	26.34 KB	-rw-r--r--

Code Editor : redisModel.py

#coding: utf-8
#-------------------------------------------------------------------
# aaPanel
#-------------------------------------------------------------------
# Copyright (c) 2015-2099 aaPanel(www.aapanel.com) All rights reserved.
#-------------------------------------------------------------------
# Author: hwliang <hwl@aapanel.com>
#-------------------------------------------------------------------

# sqlite模型
#------------------------------
import os,re,json,shutil,time
from databaseModelV2.base import databaseBase
import public
from public.validate import Param
try:
    import redis
except:
    public.ExecShell("btpip install redis")
    import redis
try:
    from BTPanel import session
except :pass

class panelRedisDB():

__DB_PASS = None
    __DB_USER = None
    __DB_PORT = 6379
    __DB_HOST = '127.0.0.1'
    __DB_CONN = None
    __DB_ERR = None

__DB_CLOUD = None
    def __init__(self):
        self.__config = self.get_options(None)

def redis_conn(self,db_idx = 0):

if self.__DB_HOST in ['127.0.0.1','localhost']:
            if not os.path.exists('/www/server/redis'): return False

if not self.__DB_CLOUD:
            self.__DB_PASS = self.__config['requirepass']
            self.__DB_PORT = int(self.__config['port'])

try:
            redis_pool = redis.ConnectionPool(host=self.__DB_HOST, port= self.__DB_PORT, password= self.__DB_PASS, db= db_idx)
            self.__DB_CONN = redis.Redis(connection_pool= redis_pool)
            return self.__DB_CONN
        except :
             self.__DB_ERR = public.get_error_info()
        return False

def set_host(self,host,port,name,username,password,prefix = ''):
        self.__DB_HOST = host
        self.__DB_PORT = int(port)
        self.__DB_NAME = name
        if self.__DB_NAME: self.__DB_NAME = str(self.__DB_NAME)
        self.__DB_USER = str(username)
        self._USER = str(username)
        self.__DB_PASS = str(password)
        self.__DB_PREFIX = prefix
        self.__DB_CLOUD = 1
        return self

#获取配置项
    def get_options(self,get = None):

result = {}
        redis_conf = public.readFile("{}/redis/redis.conf".format(public.get_setup_path()))
        if not redis_conf: return False

keys = ["bind","port","timeout","maxclients","databases","requirepass","maxmemory"]
        for k in keys:
            v = ""
            rep = "\n%s\\s+(.+)" % k
            group = re.search(rep,redis_conf)
            if not group:
                if k == "maxmemory":
                    v = "0"
                if k == "maxclients":
                    v = "10000"
                if k == "requirepass":
                    v = ""
            else:
                if k == "maxmemory":
                    v = int(group.group(1)) / 1024 / 1024
                else:
                    v = group.group(1)
            result[k] = v
        return result

class main(databaseBase):

_db_max = 16  #最大redis数据库
    def __init__(self):
        pass

def GetCloudServer(self,args):
        '''
            @name 获取远程服务器列表
            @author hwliang<2021-01-10>
            @return list
        '''
        # # 校验参数
        # try:
        #     args.validate([
        #         Param('type').Require().String('in', ['redis']),
        #     ], [
        #         public.validate.trim_filter(),
        #     ])
        # except Exception as ex:
        #     public.print_log("error info: {}".format(ex))
        #     return public.return_message(-1, 0, str(ex))
        # return self.GetBaseCloudServer(args)
        return public.return_message(0, 0,  self.GetBaseCloudServer(args))

def AddCloudServer(self,args):
        '''
        @添加远程数据库
        '''
        # {"db_host":"192.168.66.129","db_port":"6379","db_user":"root","db_password":"password1","db_ps":"192.168.66.129","type":"redis"}
        # 校验参数
        try:
            args.validate([

Param('db_host').Require(),
                Param('db_port').Require().Number(">=", 1).Number("<=", 65535),
                Param('db_user').Require().String(),
                Param('db_password').Require().String(),
                Param('db_ps').Require().String(),
                Param('type').Require().String(),
            ], [
                public.validate.trim_filter(),
            ])
        except Exception as ex:
            public.print_log("error info: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

return self.AddBaseCloudServer(args)
        # return public.return_message(0, 0, self.AddBaseCloudServer(args))

def RemoveCloudServer(self,args):
        '''
        @删除远程数据库
        '''
        return self.RemoveBaseCloudServer(args)
        # return public.return_message(0, 0, self.RemoveBaseCloudServer(args))

def ModifyCloudServer(self,args):
        '''
        @修改远程数据库
        '''
        return self.ModifyBaseCloudServer(args)

# return public.return_message(0, 0, self.ModifyBaseCloudServer(args))

def get_obj_by_sid(self,sid = 0,conn_config = None):
        """
        @取mssql数据库对像 By sid
        @sid 数据库分类，0：本地
        """
        if type(sid) == str:
            try:
                sid = int(sid)
            except :sid = 0

if sid:
            if not conn_config: conn_config = public.M('database_servers').where("id=?" ,sid).find()
            db_obj = panelRedisDB()

try:
                db_obj = db_obj.set_host(conn_config['db_host'],conn_config['db_port'],None,conn_config['db_user'],conn_config['db_password'])
            except Exception as e:
                raise public.PanelError(e)
        else:
            db_obj = panelRedisDB()
        return db_obj

def get_list(self,args):
        """
        @获取数据库列表
        @sql_type = redis
        """

# 校验参数
        try:
            args.validate([
                Param('sid').Require().Integer(),
            ], [
                public.validate.trim_filter(),
            ])
        except Exception as ex:
            public.print_log("error info: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

result = []
        self.sid = args.get('sid/d',0)
        for x in range(0,self._db_max):

data = {}
            data['id'] = x
            data['name'] = 'DB{}'.format(x)

try:
                redis_obj = self.get_obj_by_sid(self.sid).redis_conn(x)

data['keynum'] = redis_obj.dbsize()
                if data['keynum'] > 0:
                    result.append(data)
            except Exception as ex:
                public.print_log("error info2: {}".format(ex))
                pass

#result = sorted(result,key= lambda  x:x['keynum'],reverse=True)
        # return result
        return public.return_message(0, 0, result)

def set_redis_val(self,args):
        """
        @设置或修改指定值
        """
        # {"val":"bbbbbb12","endtime":"30","name":"aaaaa","db_idx":0,"sid":0}
        # 校验参数
        try:
            args.validate([

Param('db_idx').Require().Integer(),
                Param('sid').Require().Integer(),
                Param('name').Require().String(),  # 键
                Param('val').Require().String(),  # 值
                Param('endtime').Integer(),    # 过期时间
            ], [
                public.validate.trim_filter(),
            ])
        except Exception as ex:
            public.print_log("error info: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

self.sid = args.get('sid/d',0)
        if not 'name' in args or not 'val' in args:
            # return public.returnMsg(False, public.lang("Parameter passing error."))
            return public.return_message(-1, 0, public.lang("Parameter passing error"))

endtime = 0
        if 'endtime' in args : endtime = int(args.endtime)

redis_obj = self.get_obj_by_sid(self.sid).redis_conn(args.db_idx)
        if endtime:
            redis_obj.set(args.name, args.val, endtime)
        else:
            redis_obj.set(args.name, args.val)
        public.set_module_logs('linux_redis','set_redis_val',1)
        # return public.returnMsg(True, public.lang("Operation is successful."))
        return public.return_message(0, 0, public.lang("Operation is successful"))

def del_redis_val(self,args):
        """
        @删除key值
        """
        # {"db_idx":0,"key":"qq","sid":0}
        # 校验参数
        try:
            args.validate([

Param('db_idx').Require().Integer(),
                Param('sid').Require().Integer(),
                Param('key').Require().String(),  # 键
            ], [
                public.validate.trim_filter(),
            ])
        except Exception as ex:
            public.print_log("error info: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

self.sid = args.get('sid/d',0)
        if  not 'key' in args:
            # return public.returnMsg(False, public.lang("Parameter passing error."))
            return public.return_message(-1, 0, public.lang("Parameter passing error"))

redis_obj = self.get_obj_by_sid(self.sid).redis_conn(args.db_idx)
        redis_obj.delete(args.key)

# return public.returnMsg(True, public.lang("Operation is successful."))
        return public.return_message(0, 0, public.lang("Operation is successful"))

def clear_flushdb(self,args):
        """
        清空数据库
        @ids 清空数据库列表，不传则清空所有
        """
        # 校验参数
        try:
            args.validate([

Param('ids').String(),   # "ids":"[0,1]"
                Param('sid').Require().Integer(),
            ], [
                public.validate.trim_filter(),
            ])
        except Exception as ex:
            public.print_log("error info: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

self.sid = args.get('sid/d',0)
        ids = json.loads(args.ids)
        #ids = []
        if len(ids) == 0:
             for x in range(0,self._db_max):
                 ids.append(x)

for x in ids:
            redis_obj = self.get_obj_by_sid(self.sid).redis_conn(x)
            redis_obj.flushdb()

# return public.returnMsg(True, public.lang("Operation is successful."))
        return public.return_message(0, 0, public.lang("Operation is successful"))

def get_db_keylist(self,args):
        """
        @获取指定数据库key集合
        """
        # 校验参数
        try:
            args.validate([
                Param('db_type').Integer(),
                Param('db_idx').Require().Integer(),
                Param('limit').Integer(),
                Param('p').Integer(),
                Param('search').String(),
                Param('tojs').String(),    # 不知道
            ], [
                public.validate.trim_filter(),
            ])
        except Exception as ex:
            public.print_log("error info: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

search = '*'
        if 'search' in args: search = "*" + args.search+"*"
        db_idx = args.db_idx
        self.sid = args.get('sid/d',0)

redis_obj = self.get_obj_by_sid(self.sid).redis_conn(db_idx)
        try:
            keylist = sorted(redis_obj.keys(search))
        except :
            keylist = []

info = {'p':1,'row':10,'count':len(keylist)}

if hasattr(args,'limit'): info['row'] = int(args.limit)
        if hasattr(args,'p'): info['p']  = int(args['p'])

import page
        #实例化分页类
        page = page.Page()

info['uri'] = args
        info['return_js'] = ''
        if hasattr(args,'tojs'):
            info['return_js'] = args.tojs

slist = keylist[(info['p']-1) * info['row']:info['p'] * info['row']]

rdata = {}
        rdata['page'] = page.GetPage(info,'1,2,3,4,5,8')
        rdata['where'] = ''
        rdata['data'] = []

idx = 0
        for key in slist:
            item = {}
            try:
                item['name'] = key.decode()
            except:
                item['name'] = str(key)

item['endtime'] = redis_obj.ttl(key)
            if item['endtime'] == -1: item['endtime'] = 0

item['type'] = redis_obj.type(key).decode()

if item['type'] == 'string':
                try:
                    item['val'] = redis_obj.get(key).decode()
                except:
                    item['val'] = str(redis_obj.get(key))
            elif item['type'] == 'hash':
                item['val'] = str(redis_obj.hgetall(key))
            elif item['type'] == 'list':
                item['val'] = str(redis_obj.lrange(key, 0, -1))
            elif item['type'] == 'set':
                item['val'] = str(redis_obj.smembers(key))
            elif item['type'] == 'zset':
                item['val'] = str(redis_obj.zrange(key, 0, 1, withscores=True))
            else:
                item['val'] = ''
            try:
                item['len'] = redis_obj.strlen(key)
            except:
                item['len'] = len(item['val'])
            item['val'] = public.xsssec(item['val'])
            item['name'] = public.xsssec(item['name'])
            rdata['data'].append(item)
            idx += 1
        # return rdata
        return public.return_message(0, 0, rdata)

def ToBackup(self,args):
        """
        @备份数据库
        """

redis_obj = self.get_obj_by_sid(self.sid).redis_conn(0)
            redis_obj.save()

src_path = '{}/dump.rdb'.format(redis_obj.config_get()['dir'])
            if not os.path.exists(src_path):
                # return public.returnMsg(False, public.lang("Backup error"))
                return public.return_message(-1, 0, public.lang("Backup error"))

backup_path = session['config']['backup_path'] + '/database/redis/'
            if not os.path.exists(backup_path): os.makedirs(backup_path)

fileName = backup_path + str(self.sid) + '_db_' + time.strftime('%Y%m%d_%H%M%S',time.localtime()) +'.rdb'

shutil.copyfile(src_path,fileName)
            if not os.path.exists(fileName):
                # return public.returnMsg(False, public.lang("Backup error"))
                return public.return_message(-1, 0, public.lang("Backup error"))

# return public.returnMsg(True, public.lang("Backup Succeeded!"))
            return public.return_message(0, 0, public.lang("Backup Succeeded!"))

except Exception as ex:
            public.print_log("error info22: {}".format(ex))
            return public.return_message(-1, 0, str(ex))
    def DelBackup(self,args):
        """
        @删除备份文件
        """
        # 校验参数
        try:
            args.validate([
                Param('file').Require().SafePath(),
                # Param('sid').Require().Integer(),
            ], [
                public.validate.trim_filter(),
            ])
        except Exception as ex:
            public.print_log("error info: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

file = args.file

if os.path.exists(file):
            os.remove(file)

# return public.returnMsg(True, public.lang("Delete successfully!"))
        return public.return_message(0, 0, public.lang("Delete successfully!"))

def InputSql(self,get):
        """
        @导入数据库
        """

# 校验参数
        try:
            get.validate([
                Param('file').Require().SafePath(),
                Param('sid').Require().Integer(),
            ], [
                public.validate.trim_filter(),
            ])
        except Exception as ex:
            public.print_log("error info: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

file = get.file
        self.sid = get.get('sid/d',0)

redis_obj = self.get_obj_by_sid(self.sid).redis_conn(0)

rpath = redis_obj.config_get()['dir']
        dst_path = '{}/dump.rdb'.format(rpath)
        public.ExecShell("/etc/init.d/redis stop")
        if os.path.exists(dst_path): os.remove(dst_path)
        shutil.copy2(file, dst_path)
        public.ExecShell("chown redis.redis {dump} && chmod 644 {dump}".format(dump=dst_path))
        # self.restart_services()
        public.ExecShell("/etc/init.d/redis start")
        if os.path.exists(dst_path):
            # return public.returnMsg(True, public.lang("Restore Successful."))
            return public.return_message(0, 0, public.lang("Restore Successful."))
        # return public.returnMsg(False, public.lang("Restore failure."))
        return public.return_message(-1, 0, public.lang("Restore failure."))

@staticmethod
    def _get_backup(search: str, files_path_list: list, cloud_list: dict, current_path: str) -> list:
        """
        获取指定目录下redis备份文件列表
        """
        res = []
        for file_name in files_path_list:
            try:
                if search:
                    if file_name.lower().find(search) == -1:
                        continue
                arrs = file_name.split('_')
                filepath = '{}/{}'.format(current_path, file_name).replace('//', '/')
                stat = os.stat(filepath)
                item = {
                    'name': file_name,
                    'filepath': filepath,
                    'size': stat.st_size,
                    'mtime': int(stat.st_mtime),
                    'sid': arrs[0],
                    'conn_config': cloud_list.get(f"id-{str(arrs[0])}", {}),
                }
                res.append(item)
            except Exception as ex:
                public.print_log("error info: {}".format(ex))
                continue
        return res

def get_backup_list(self,get):
        """
        @获取备份文件列表
        """
        # 校验参数
        try:
            get.validate([
                Param('search').String(),
            ], [
                public.validate.trim_filter(),
            ])
        except Exception as ex:
            public.print_log("error info: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

search = ''
        if hasattr(get, 'search'):
            search = get['search'].strip().lower()

nlist = []
        cloud_list = {}
        listm = self.GetCloudServer({'type': 'redis'})

for x in listm['message']:
            cloud_list['id-' + str(x['id'])] = x

path = session['config']['backup_path'] + '/database/redis/'
        if not os.path.exists(path):
            os.makedirs(path)

all_path = os.listdir(path)
        if 'crontab_backup' in all_path:
            cron_path_list = os.listdir(os.path.join(path, 'crontab_backup'))
            nlist.extend(self._get_backup(
                search=search,
                files_path_list=cron_path_list,
                cloud_list=cloud_list,
                current_path=os.path.join(path, 'crontab_backup'),
            ))
            all_path.remove('crontab_backup')

nlist.extend(self._get_backup(
            search=search,
            files_path_list=all_path,
            cloud_list=cloud_list,
            current_path=path,
        ))

if hasattr(get, 'sort'):
            nlist = sorted(nlist, key=lambda data: data['mtime'], reverse=get["sort"] == "desc")

return public.return_message(0, 0,  nlist)

def restart_services(self):
        """
        @重启服务
        """
        public.ExecShell('net stop redis')
        public.ExecShell('net start redis')
        return True

def check_cloud_database_status(self,conn_config):
        """
        @检测远程数据库是否连接
        @conn_config 远程数据库配置，包含host port pwd等信息
        """
        try:

sql_obj = panelRedisDB().set_host(conn_config['db_host'],conn_config['db_port'],conn_config['db_name'],conn_config['db_user'],conn_config['db_password'])
            keynum = sql_obj.redis_conn(0).dbsize()
            # return True
            return public.return_message(0, 0, True)
        except Exception as ex:

# return public.returnMsg(False,ex)
            return public.return_message(-1, 0, ex)

${this.title}

Code Editor : redisModel.py