1 Star 0 Fork 3

heisebaijiu/DBAS

forked from 春秋/DBAS 
加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
克隆/下载
commands.py 7.10 KB
一键复制 编辑 原始数据 按行查看 历史
春秋 提交于 2015-09-09 16:16 . DBAS
#!/usr/bin/env python3
# coding:utf-8
"""
DAS指令集
"""
__author__ = 'MC'
import re
import sys
import traceback
import logging
import loadconfig
import backupdb
import config
import cerror
# 检查系统环境并载入对应模块
if loadconfig.System == 'win32':
from WinModule import tasks
from WinModule import display
class Command:
def __init__(self):
self.ShowDict = {'DbList': True, 'TkList': True, 'SysConfig': True}
self.TK = tasks.Task()
self.CO = config.Config()
# 配置logging的默认格式
logging.basicConfig(format="%(asctime)s [%(levelname)s] [%(filename)s] %(message)s",
filename=loadconfig.LogPath + "/Error.log")
def Init(self):
#可用命令列表及对应规则集
self.Command = {
'Backup': (self.Backup, re.compile(r"(?P<Command>Backup) (?P<LinkName>\S*)(?P<FilePath>.*)")),
'Help': (self.Help, re.compile(r"(?P<Command>Help) (?P<Action>.+)")),
'Show': (self.Show, re.compile(r"(?P<Command>Show) (?P<Action>DbList|TkList|SysConfig)")),
'Config': (self.Config, re.compile(r"(?P<Command>Config) (?P<Action>Database|Task) (?P<Name>.+?):(?P<Value>.*)")),
'Del': (self.Del, re.compile(r"(?P<Command>Del) (?P<Action>Task|Database) (?P<Name>.+)")),
'Test': (self.Test, re.compile(r"(?P<Command>Test) (?P<Name>.+)")),
'Sys': (self.Sys, re.compile(r"(?P<Command>Sys) (?P<Action>BPath) (?P<Value>.+)"))
}
def Backup(self,**Command):
try:
Link = self.CO.Show_Db_Link(Command['LinkName'])
if Link:
backupdb.BackupMyDb(Host=Link['host'], User=Link['user'], Password=Link['password'], DbName=Link['database'], Encoding=Link['encode'], Port=Link['port']).Backup('None' if Command['FilePath'] == '' else Command['FilePath'])
else:
print("不存在这个实例链接,请使用:Show DbList查询已定义的链接")
except Exception:
# print("系统错误:请查看程序目录下/Log/Error.log文件")
E_Type, E_Value, E_TB = sys.exc_info()
Trace_FileName, Trace_LineNo, Trace_FunName, Trace_Source = traceback.extract_tb(E_TB)[0]
logging.error("[File:{0}] [LineNo:{1}] [Mes:{2}] [Source:{3}]".format(Trace_FileName, Trace_LineNo, E_Value, Trace_Source))
def Help(self, Command):
try:
if Command == 'List':
display.Dis_Command({'配置命令': ['Config', 'Del', 'Sys'], '查看信息命令': ['Show'],
'数据库命令': ['Backup', 'Test'], '计划任务命令': ['Task']})
if Command == 'Show':
display.Dis_Help_Show()
if Command == "Backup":
display.Dis_Help_Bakcup()
if Command == "Config":
display.Dis_Help_Config()
if Command == "Sys":
display.Dis_Help_Sys()
except:
print("指令错误,请使用:Help <Command>获取指令帮助,如需获取指令列表,请使用:Help List")
def Show(self, Command):
if self.ShowDict.get(Command,False):
if Command == 'DbList':
self.CO.Read_Db_Config()
if Command == 'TkList':
self.CO.Read_Tk_Config()
if Command == 'SysConfig':
self.CO.Read_Sys()
else:
print("指令不存在,请使用:Help Show获取指令帮助")
def Config(self, **Command):
try:
#解析指令的Value部分,如果为空则抛出异常
ParamArr = Command['Value'].split(',')
if len(ParamArr) < 1: raise cerror.CommandError("参数格式错误,如需详细帮助,请使用Help Config命令")
Params = {}
for Param in ParamArr:
#解析Value部分中的具体参数对,如果参数对不符合格式,则抛出异常
List = Param.split('=')
if len(List) < 2: raise cerror.CommandError("参数格式错误,如需详细帮助,请使用Help Config命令")
Params[List[0]] = List[1]
if Command['Action'] == "Database":
self.CO.Set_Db_Link(Command['Name'], Params.get('host', ''), Params.get('port', '3306'), Params.get('user', ''), Params.get('password', ''), Params.get('database', ''), Params.get('encode', 'utf8'))
if Command['Action'] == "Task":
self.CO.Set_Tk_Config(Command['Name'], Params.get('sc', 'WEEKLY'), Params.get('mo', '1'), Params.get('d', 'Mon'), Params.get('st', '11:00'), Params.get('dl', ''))
except cerror.CommandError as E:
print(E.Message)
except Exception:
print("系统错误:请查看程序目录下/Log/Error.log文件")
E_Type, E_Value, E_TB = sys.exc_info()
Trace_FileName, Trace_LineNo, Trace_FunName, Trace_Source = traceback.extract_tb(E_TB)[0]
logging.error("[File:{0}] [LineNo:{1}] [Mes:{2}] [Source:{3}]".format(Trace_FileName, Trace_LineNo, E_Value, Trace_Source))
def Del(self, **Command):
try:
if Command.get('Action', '') == "Task":
Task = self.CO.Show_Tk_Info(Command.get('Name', ''))
if Task:
self.CO.Del_Tk(Task.get('name', ''))
self.TK.Delete_Task(Task.get('name', ''))
else:
print("不存在这个任务,请使用Show TkList查看添加的任务")
elif Command.get('Action', '') == "Database":
Database = self.CO.Show_Db_Link(Command.get('Name', ''))
if Database:
self.CO.Del_Db(Command['Name'])
else:
print("参数错误,如需帮助,请使用:Help Del")
except Exception:
print("系统错误:请查看程序目录下/Log/Error.log文件")
E_Type, E_Value, E_TB = sys.exc_info()
Trace_FileName, Trace_LineNo, Trace_FunName, Trace_Source = traceback.extract_tb(E_TB)[0]
logging.error("[File:{0}] [LineNo:{1}] [Mes:{2}] [Source:{3}]".format(Trace_FileName, Trace_LineNo, E_Value, Trace_Source))
def Test(self, **Command):
try:
Link = self.CO.Show_Db_Link(Command['Name'])
if Link:
backupdb.BackupMyDb(Host=Link['host'], User=Link['user'], Password=Link['password'], DbName=Link['database'], Encoding=Link['encode'], Port=Link['port']).Test()
else:
print("不存在这个实例链接,请使用:Show DbList查询已定义的链接")
except Exception:
print("系统错误:请查看程序目录下/Log/Error.log文件")
E_Type, E_Value, E_TB = sys.exc_info()
Trace_FileName, Trace_LineNo, Trace_FunName, Trace_Source = traceback.extract_tb(E_TB)[0]
logging.error("[File:{0}] [LineNo:{1}] [Mes:{2}] [Source:{3}]".format(Trace_FileName, Trace_LineNo, E_Value, Trace_Source))
def Sys(self, **Command):
if Command.get('Action', '') != "":
self.CO.Set_Sys(Command['Action'], Command['Value'])
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
Python
1
https://gitee.com/heisebaijiu/DBAS.git
git@gitee.com:heisebaijiu/DBAS.git
heisebaijiu
DBAS
DBAS
master

搜索帮助

0d507c66 1850385 C8b1a773 1850385