forked from aaPanel/BaoTa
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathpanelMysql.py
134 lines (121 loc) · 4.64 KB
/
panelMysql.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
#coding: utf-8
# +-------------------------------------------------------------------
# | 宝塔Linux面板
# +-------------------------------------------------------------------
# | Copyright (c) 2015-2099 宝塔软件(http://bt.cn) All rights reserved.
# +-------------------------------------------------------------------
# | Author: hwliang <[email protected]>
# +-------------------------------------------------------------------
import re,os,sys,public
class panelMysql:
__DB_PASS = None
__DB_USER = 'root'
__DB_PORT = 3306
__DB_HOST = 'localhost'
__DB_CONN = None
__DB_CUR = None
__DB_ERR = None
__DB_NET = None
#连接MYSQL数据库
def __Conn(self):
if self.__DB_NET: return True
try:
myconf = public.readFile('/etc/my.cnf')
socket_re = re.search(r"socket\s*=\s*(.+)",myconf)
if socket_re:
socket = socket_re.groups()[0]
else:
socket = '/tmp/mysql.sock'
try:
if sys.version_info[0] != 2:
try:
import pymysql
except:
public.ExecShell("pip install pymysql")
import pymysql
pymysql.install_as_MySQLdb()
import MySQLdb
if sys.version_info[0] == 2:
reload(MySQLdb)
except:
try:
import pymysql
pymysql.install_as_MySQLdb()
import MySQLdb
except Exception as e:
self.__DB_ERR = e
return False
try:
rep = r"port\s*=\s*([0-9]+)"
self.__DB_PORT = int(re.search(rep,myconf).groups()[0])
except:
self.__DB_PORT = 3306
self.__DB_PASS = public.M('config').where('id=?',(1,)).getField('mysql_root')
try:
self.__DB_CONN = MySQLdb.connect(host = self.__DB_HOST,user = self.__DB_USER,passwd = self.__DB_PASS,port = self.__DB_PORT,charset="utf8",connect_timeout=1,unix_socket=socket)
except MySQLdb.Error as e:
self.__DB_HOST = '127.0.0.1'
self.__DB_CONN = MySQLdb.connect(host = self.__DB_HOST,user = self.__DB_USER,passwd = self.__DB_PASS,port = self.__DB_PORT,charset="utf8",connect_timeout=1,unix_socket=socket)
self.__DB_CUR = self.__DB_CONN.cursor()
return True
except MySQLdb.Error as e:
self.__DB_ERR = e
return False
#连接远程数据库
def connect_network(self,host,port,username,password):
self.__DB_NET = True
try:
try:
if sys.version_info[0] != 2:
try:
import pymysql
except:
public.ExecShell("pip install pymysql")
import pymysql
pymysql.install_as_MySQLdb()
import MySQLdb
if sys.version_info[0] == 2:
reload(MySQLdb)
except:
try:
import pymysql
pymysql.install_as_MySQLdb()
import MySQLdb
except Exception as e:
self.__DB_ERR = e
return False
self.__DB_CONN = MySQLdb.connect(host = host,user = username,passwd = password,port = port,charset="utf8",connect_timeout=10)
pymysql.connect
self.__DB_CUR = self.__DB_CONN.cursor()
except MySQLdb.Error as e:
self.__DB_ERR = e
return False
def execute(self,sql):
#执行SQL语句返回受影响行
if not self.__Conn(): return self.__DB_ERR
try:
result = self.__DB_CUR.execute(sql)
self.__DB_CONN.commit()
self.__Close()
return result
except Exception as ex:
return ex
def query(self,sql):
#执行SQL语句返回数据集
if not self.__Conn(): return self.__DB_ERR
try:
self.__DB_CUR.execute(sql)
result = self.__DB_CUR.fetchall()
#将元组转换成列表
if sys.version_info[0] == 2:
data = map(list,result)
else:
data = list(map(list,result))
self.__Close()
return data
except Exception as ex:
return ex
#关闭连接
def __Close(self):
self.__DB_CUR.close()
self.__DB_CONN.close()