1 Star 0 Fork 3

heisebaijiu/DBAS

forked from 春秋/DBAS 
加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
克隆/下载
config.py 11.69 KB
一键复制 编辑 原始数据 按行查看 历史
春秋 提交于 2015-09-09 16:16 . DBAS
#!/usr/bin/env python3
# coding:utf-8
__author__ = 'MC'
import os
import sys
import traceback
import logging
import time
import xml.etree.ElementTree as Etree
import loadconfig
# 检查系统环境并载入对应模块
if loadconfig.System == 'win32':
from WinModule import tasks
from WinModule import display
class Config:
"""
配置项操作类
"""
def __init__(self):
self.DbConfig = loadconfig.ConfigPath + "/DbConfig.xml"
self.TkConfig = loadconfig.ConfigPath + "/Tasks.xml"
self.System = loadconfig.ConfigPath + "/Config.xml"
logging.basicConfig(format="%(asctime)s [%(levelname)s] [%(filename)s] %(message)s",
filename=loadconfig.LogPath + "/Error.log")
def Read_Db_Config(self):
"""
读取数据库连接配置信息
:return:
"""
try:
Tree = Etree.parse(self.DbConfig)
Root = Tree.getroot()
Lines = Root.findall('line')
for Line in Lines:
display.Dis_Db(Line)
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 Set_Db_Link(self, LinkName, Host, Port, User, Password, Database, Encode):
"""
设置数据库连接
:param LinkName: 连接的标识名称
:param Host: 连接地址
:param Port: 连接端口
:param User: 用户名
:param Password: 密码
:param Database: 数据库名
:param Encode: 编码
:return:
"""
try:
Tree = Etree.parse(self.DbConfig)
Root = Tree.getroot()
Lines = Root.findall('line')
LinkNames = [Line.attrib['name'] for Line in Lines]
if LinkName not in LinkNames:
Line = Etree.SubElement(Root,'line',{'name': LinkName})
Etree.SubElement(Line, 'host').text = Host
Etree.SubElement(Line, 'port').text = Port
Etree.SubElement(Line, 'user').text = User
Etree.SubElement(Line, 'password').text = Password
Etree.SubElement(Line, 'database').text = Database
Etree.SubElement(Line, 'encode').text = Encode
Etree.SubElement(Line, 'time').text = str(time.time())
Etree.SubElement(Line, 'u_time').text = str(time.time())
Tree.write(self.DbConfig,'utf-8')
print("添加实例链接成功")
else:
Link = [Line for Line in Lines if Line.attrib['name'] == LinkName]
Link[0].find('host').text = Host or Link[0].find('host').text
Link[0].find('port').text = Port or Link[0].find('port').text
Link[0].find('user').text = User or Link[0].find('user').text
Link[0].find('password').text = Password or Link[0].find('password').text
Link[0].find('database').text = Database or Link[0].find('database').text
Link[0].find('encode').text = Encode or Link[0].find('encode').text
Link[0].find('u_time').text = int(time.time())
Tree.write(self.DbConfig, 'utf-8')
print("更新实例链接成功")
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 Show_Db_Link(self, LinkName):
"""
查询指定连接名称的详细数据
:param LinkName:
:return:
"""
try:
Tree = Etree.parse(self.DbConfig)
Root = Tree.getroot()
Link = [Line for Line in Root.findall('line') if Line.attrib['name'] == LinkName]
if len(Link) > 0:
return {'host': Link[0].find('host').text, 'port': Link[0].find('port').text, 'user': Link[0].find('user').text, 'password': Link[0].find('password').text, 'database': Link[0].find('database').text, 'encode': Link[0].find('encode').text}
else:
return False
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_Db(self,LinkName):
try:
Tree = Etree.parse(self.DbConfig)
Root = Tree.getroot()
Link = [Link for Link in Root.findall("line") if Link.attrib['name'] == LinkName]
if Link:
Root.remove(Link[0])
Tree.write(self.DbConfig, 'utf-8')
print("删除数据库连接 {0} 成功".format(LinkName))
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 Read_Tk_Config(self):
"""
读取已配置的计划任务信息
:return:
"""
try:
Tree = Etree.parse(self.TkConfig)
Root = Tree.getroot()
Tasks = Root.findall('task')
for Task in Tasks:
display.Dis_Tk(Task)
if Task == Tasks[-1]:
break
else:
print("当前没有计划任务")
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 Set_Tk_Config(self, TkName, SC, MO, D, ST, DL):
try:
Tree = Etree.parse(self.TkConfig)
Root = Tree.getroot()
TaskList = [Task.attrib['name'] for Task in Root.findall("task")]
if TkName not in TaskList:
Task = Etree.SubElement(Root,"task", {'name': TkName})
Etree.SubElement(Task,"name").text = TkName
Etree.SubElement(Task,"run").text = "E:\python\pyobj\DAS\TaskRun.pyw -t {0}".format(TkName)
Etree.SubElement(Task,"sc").text = SC
Etree.SubElement(Task,"mo").text = MO
Etree.SubElement(Task,"d").text = D
Etree.SubElement(Task,"st").text = ST
Etree.SubElement(Task,"dl").text = DL
Etree.SubElement(Task, 'time').text = str(time.time())
Etree.SubElement(Task, 'u_time').text = str(time.time())
Tree.write(self.TkConfig, 'utf-8')
print("创建计划任务 {0} 成功".format(TkName))
tasks.Task().Create_Task(TkName, "E:\python\pyobj\DAS\TaskRun.pyw -t {0}".format(TkName), SC, MO, D, ST)
else:
Task = [Task for Task in Root.findall("task") if Task.attrib['name'] == TkName]
Task[0].find("sc").text = SC or Task[0].find("sc").text
Task[0].find("mo").text = MO or Task[0].find("mo").text
Task[0].find("d").text = D or Task[0].find("d").text
Task[0].find("st").text = ST or Task[0].find("st").text
Task[0].find("dl").text = DL or Task[0].find("dl").text
Task[0].find("u_time").text = str(time.time())
Tree.write(self.TkConfig, 'utf-8')
print("更新计划任务 {0} 成功".format(TkName))
tasks.Task().Change_Task(TkName, ST)
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 Show_Tk_Info(self, TaskName):
try:
Tree = Etree.parse(self.TkConfig)
Root = Tree.getroot()
Task = [Task for Task in Root.findall('task') if Task.attrib['name'] == TaskName]
if len(Task) > 0:
return {'name': Task[0].find('name').text, 'run': Task[0].find('run').text, 'sc':Task[0].find('sc').text, 'mo': Task[0].find('mo').text, 'd': Task[0].find('d').text, 'st': Task[0].find('st').text, 'dl': Task[0].find('dl').text}
else:
return False
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_Tk(self,TaskName):
try:
Tree = Etree.parse(self.TkConfig)
Root = Tree.getroot()
Task = [Task for Task in Root.findall('task') if Task.attrib['name'] == TaskName]
if len(Task) > 0:
Root.remove(Task[0])
Tree.write(self.TkConfig, 'utf-8')
print("删除计划任务 {0} 成功".format(TaskName))
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 Set_Sys(self, Option, Value):
try:
Tree = Etree.parse(self.System)
Root = Tree.getroot()
if Option == "BPath":
if os.path.isdir(Value):
Root.find('backuppath').text = Value
Tree.write(self.System, 'utf-8')
print("设置备份目录成功")
else:
print("设置的值不是一个目录,请重新设定")
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 Read_Sys(self):
Tree = Etree.parse(self.System)
Root = Tree.getroot()
SysDict = {Sys.tag:Sys.text for Sys in Root}
display.Dis_Sys(SysDict)
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
Python
1
https://gitee.com/heisebaijiu/DBAS.git
git@gitee.com:heisebaijiu/DBAS.git
heisebaijiu
DBAS
DBAS
master

搜索帮助

0d507c66 1850385 C8b1a773 1850385