#!/usr/bin/python
# -*- coding: cp936 -*-
#################################
# written by caocao #
# [email protected] #
# http://nethermit.yeah.net #
#################################
import sys, string, os
from types import *
import mysqldb
print "written by caocao"
print "[email protected]"
print "http://nethermit.yeah.net"
print
def iif(expression, whentrue, whenfalse):
if expression:
return whentrue
else:
return whenfalse
class mysqltest:
def __init__(self, host="localhost", user="root", passwd="", db=""):
self.connection=none
self.host=host
self.user=user
self.passwd=passwd
self.db=db
self.result=none
print "-"*40
print "mysql shell v 1.0"
print "usage: python mysql.shell.py [host] [user] [passwd(% is empty)] [db]"
print "connect..."
try:
self.connection=mysqldb.connect(host=self.host, user=self.user, passwd=self.passwd, db=self.db)
except:
print "can't connect to mysql server./nplease make sure your username or password is right."
sys.exit(1)
print "-"*40
print self.printcomment("connection", "get_server_info")
print self.printcomment("connection", "get_host_info")
print self.printcomment("connection", "get_proto_info")
print self.printcomment("connection", "info")
print self.printcomment("connection", "character_set_name")
print self.printcomment("connection", "thread_id")
print self.printcomment("connection", "stat")
def __del__(self):
if self.connection!=none:
self.connection.close()
print "-"*40
print "quit..."
def printcomment(self, instance, function):
return "%s = %s" % (string.rjust(function, 18), eval("self."+instance+"."+function+"()"))
def printall(self):
output, row="", self.result.fetch_row(0)
for i in range(self.result.num_fields()):
output+=repr(self.result.describe()[i][0])+"/n"
for i in range(self.result.num_rows()):
for j in range(self.result.num_fields()):
output+=iif(type(row[i][j]) is stringtype, row[i][j], repr(row[i][j]))+"/n"
return output
def runsql(self, querystring="show databases"):
print "-"*40
try:
self.connection.query(querystring)
except:
print "can't run sql."
else:
self.result=self.connection.store_result()
print self.printcomment("connection", "field_count")
print self.printcomment("connection", "affected_rows")
print self.printcomment("connection", "insert_id")
print self.printcomment("result", "num_fields")
print self.printcomment("result", "num_rows")
print self.printcomment("result", "field_flags")
print "-"*40
print self.printall()
if __name__=="__main__":
argarray=sys.argv
del argarray[0]
test=eval(("mysqltest(/"%s/")" % "/",/"".join(argarray)).replace("%", ""))
while true:
try:
command=string.strip(raw_input("ps mysql>"), " ")
commandlow=string.lower(command)
except eoferror:
break
else:
if commandlow=="exit" or commandlow=="quit":
break
elif commandlow=="":
continue
else:
test.runsql(command)
