126 lines
2.9 KiB
Python
126 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
PMS 数据库工具
|
|
用法：
|
|
python tools/db.py import <sql文件> # 导入SQL文件
|
|
python tools/db.py query <SQL语句> # 执行查询
|
|
python tools/db.py exec <SQL语句> # 执行写入
|
|
python tools/db.py tables # 列出所有表
|
|
python tools/db.py count # 统计各表数据量
|
|
"""
|
|
|
|
import sys
|
|
import pymysql
|
|
|
|
# 数据库配置
|
|
DB_CONFIG = {
|
|
'host': '10.8.0.252',
|
|
'user': 'root',
|
|
'password': 'Hbhyg731024@',
|
|
'database': 'pms',
|
|
'charset': 'utf8mb4',
|
|
}
|
|
|
|
def get_conn():
|
|
return pymysql.connect(**DB_CONFIG)
|
|
|
|
def import_sql(filepath):
|
|
"""导入SQL文件"""
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
sql = f.read()
|
|
|
|
conn = get_conn()
|
|
cur = conn.cursor()
|
|
|
|
# 按分号分割SQL语句
|
|
statements = [s.strip() for s in sql.split(';') if s.strip()]
|
|
|
|
success = 0
|
|
fail = 0
|
|
for stmt in statements:
|
|
try:
|
|
cur.execute(stmt)
|
|
success += 1
|
|
except Exception as e:
|
|
print(f" ✗ 失败: {str(e)[:80]}")
|
|
fail += 1
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
print(f"导入完成: 成功 {success}, 失败 {fail}")
|
|
|
|
def query(sql):
|
|
"""执行查询"""
|
|
conn = get_conn()
|
|
cur = conn.cursor()
|
|
cur.execute(sql)
|
|
columns = [desc[0] for desc in cur.description]
|
|
rows = cur.fetchall()
|
|
conn.close()
|
|
|
|
if not rows:
|
|
print("无数据")
|
|
return
|
|
|
|
# 打印表头
|
|
print('\t'.join(columns))
|
|
print('-' * 50)
|
|
for row in rows:
|
|
print('\t'.join(str(v) for v in row))
|
|
|
|
def execute(sql):
|
|
"""执行写入"""
|
|
conn = get_conn()
|
|
cur = conn.cursor()
|
|
try:
|
|
cur.execute(sql)
|
|
conn.commit()
|
|
print(f"执行成功, 影响行数: {cur.rowcount}")
|
|
except Exception as e:
|
|
print(f"执行失败: {e}")
|
|
conn.rollback()
|
|
finally:
|
|
conn.close()
|
|
|
|
def tables():
|
|
"""列出所有表"""
|
|
query("SHOW TABLES")
|
|
|
|
def count():
|
|
"""统计各表数据量"""
|
|
conn = get_conn()
|
|
cur = conn.cursor()
|
|
cur.execute("SHOW TABLES")
|
|
tables = [row[0] for row in cur.fetchall()]
|
|
|
|
print("表名\t\t数据量")
|
|
print("-" * 30)
|
|
for t in tables:
|
|
cur.execute(f"SELECT COUNT(*) FROM `{t}`")
|
|
cnt = cur.fetchone()[0]
|
|
print(f"{t}\t{cnt}")
|
|
conn.close()
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print(__doc__)
|
|
return
|
|
|
|
cmd = sys.argv[1]
|
|
|
|
if cmd == 'import' and len(sys.argv) >= 3:
|
|
import_sql(sys.argv[2])
|
|
elif cmd == 'query' and len(sys.argv) >= 3:
|
|
query(' '.join(sys.argv[2:]))
|
|
elif cmd == 'exec' and len(sys.argv) >= 3:
|
|
execute(' '.join(sys.argv[2:]))
|
|
elif cmd == 'tables':
|
|
tables()
|
|
elif cmd == 'count':
|
|
count()
|
|
else:
|
|
print(__doc__)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|