90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
通用 MySQL 查询脚本
|
|
用法：
|
|
python scripts/query_mysql.py "SELECT * FROM diagnosis_session ORDER BY created_at DESC LIMIT 5"
|
|
python scripts/query_mysql.py # 交互模式
|
|
依赖：pip install pymysql
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
# Windows 控制台 UTF-8 输出
|
|
if sys.stdout.encoding and sys.stdout.encoding.lower() != 'utf-8':
|
|
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
|
|
|
|
try:
|
|
import pymysql
|
|
import pymysql.cursors
|
|
except ImportError:
|
|
print("缺少依赖，请先执行: pip install pymysql")
|
|
sys.exit(1)
|
|
|
|
def required_env(name: str) -> str:
|
|
value = os.getenv(name)
|
|
if value is None or not value.strip():
|
|
print(f"缺少必需环境变量: {name}")
|
|
sys.exit(2)
|
|
return value.strip()
|
|
|
|
|
|
DB_CONFIG = {
|
|
"host": os.getenv("SUPERBIZ_MYSQL_HOST", "119.29.78.52"),
|
|
"port": int(os.getenv("SUPERBIZ_MYSQL_PORT", "33306")),
|
|
"user": os.getenv("SUPERBIZ_MYSQL_USERNAME", "root"),
|
|
"password": required_env("SUPERBIZ_MYSQL_PASSWORD"),
|
|
"database": os.getenv("SUPERBIZ_MYSQL_DATABASE", "superbiz_agent"),
|
|
"charset": "utf8mb4",
|
|
"cursorclass": pymysql.cursors.DictCursor,
|
|
}
|
|
|
|
|
|
def run_query(sql: str):
|
|
conn = pymysql.connect(**DB_CONFIG)
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql)
|
|
if sql.strip().upper().startswith("SELECT") or sql.strip().upper().startswith("SHOW"):
|
|
rows = cur.fetchall()
|
|
if not rows:
|
|
print("(空结果)")
|
|
return
|
|
# 打印列头
|
|
cols = list(rows[0].keys())
|
|
col_widths = {c: max(len(c), max(len(str(r[c])) for r in rows)) for c in cols}
|
|
header = " | ".join(c.ljust(col_widths[c]) for c in cols)
|
|
print(header)
|
|
print("-" * len(header))
|
|
for row in rows:
|
|
print(" | ".join(str(row[c]).ljust(col_widths[c]) for c in cols))
|
|
print(f"\n({len(rows)} 行)")
|
|
else:
|
|
conn.commit()
|
|
print(f"OK，影响行数: {cur.rowcount}")
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) > 1:
|
|
sql = " ".join(sys.argv[1:])
|
|
run_query(sql)
|
|
else:
|
|
print("MySQL 交互模式（输入 exit 退出）")
|
|
print(f"连接：{DB_CONFIG['user']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}")
|
|
print("-" * 50)
|
|
while True:
|
|
try:
|
|
sql = input("sql> ").strip()
|
|
if sql.lower() in ("exit", "quit", "q"):
|
|
break
|
|
if not sql:
|
|
continue
|
|
run_query(sql)
|
|
except KeyboardInterrupt:
|
|
break
|
|
except Exception as e:
|
|
print(f"错误: {e}")
|