Python3如何實(shí)現(xiàn)自動(dòng)生成MySQL數(shù)據(jù)字典的markdown文本

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
自動(dòng)生成MySQL數(shù)據(jù)表的數(shù)據(jù)字典支持多個(gè)
自動(dòng)獲取數(shù)據(jù)庫連接信息，方便多用
author: gxcuizy
date: 2020-04-30
"""

import pymysql
import os
import time


class DataDict(object):
 def __init__(self, connect_info):
 # 數(shù)據(jù)庫連接配置
 self.host_name = connect_info[0]
 self.user_name = connect_info[1]
 self.pwd = connect_info[2]
 self.db_name = connect_info[3]
 self.folder_name = 'mysql_dict'

 def run(self, table_str):
 """腳本執(zhí)行入口"""
 try:
  # 創(chuàng)建一個(gè)連接
  conn = pymysql.connect(self.host_name, self.user_name, self.pwd, self.db_name)
  # 用cursor()創(chuàng)建一個(gè)游標(biāo)對(duì)象
  cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
 except Exception:
  print('數(shù)據(jù)庫連接失敗，請(qǐng)檢查連接信息！')
  exit(1)
 table_list = table_str.split(',')
 for table_name in table_list:
  # 判斷表是否存在
  sql = "SHOW TABLES LIKE '%s'" % (table_name,)
  cursor.execute(sql)
  result_count = cursor.rowcount
  if result_count == 0:
  print('%s數(shù)據(jù)庫中%s表名不存在，無法生成……' % (self.db_name, table_name))
  continue
  # 表注釋獲取
  print('開始生成表%s的數(shù)據(jù)字典' % (table_name,))
  sql = "show table status WHERE Name = '%s'" % (table_name,)
  cursor.execute(sql)
  result = cursor.fetchone()
  table_comment = result['Comment']
  # 文件夾和文件處理
  file_path = self.folder_name + os.sep + table_name + '.md'
  self.deal_file(file_path)
  # 打開文件，準(zhǔn)備寫入
  dict_file = open(file_path, 'a', encoding='UTF-8')
  dict_file.write('#### %s %s' % (table_name, table_comment))
  dict_file.write('\n | 字段名稱 | 字段類型 | 默認(rèn)值 | 字段注釋 |')
  dict_file.write('\n | --- | --- | --- | --- |')
  # 表結(jié)構(gòu)查詢
  field_str = "COLUMN_NAME,COLUMN_TYPE,COLUMN_DEFAULT,COLUMN_COMMENT"
  sql = "select %s from information_schema.COLUMNS where table_schema='%s' and table_name='%s'" % (field_str, self.db_name, table_name)
  cursor.execute(sql)
  fields = cursor.fetchall()
  for field in fields:
  column_name = field['COLUMN_NAME']
  column_type = field['COLUMN_TYPE']
  column_default = str(field['COLUMN_DEFAULT'])
  column_comment = field['COLUMN_COMMENT']
  info = ' | ' + column_name + ' | ' + column_type + ' | ' + column_default + ' | ' + column_comment + ' | '
  dict_file.write('\n ' + info)
  # 關(guān)閉連接
  print('完成表%s的數(shù)據(jù)字典' % (table_name,))
  dict_file.close()
 cursor.close()
 conn.close()

 def deal_file(self, file_name):
 """處理存儲(chǔ)文件夾和文件"""
 # 不存在則創(chuàng)建文件夾
 if not os.path.exists(self.folder_name):
  os.mkdir(self.folder_name)
 # 刪除已存在的文件
 if os.path.isfile(file_name):
  os.unlink(file_name)

 def test_conn(self, conn_info):
 """測試數(shù)據(jù)庫連接"""
 try:
  # 創(chuàng)建一個(gè)連接
  pymysql.connect(conn_info[0], conn_info[1], conn_info[2], conn_info[3])
  return True
 except Exception:
  return False


# 程序執(zhí)行入口
if __name__ == '__main__':
 # 數(shù)據(jù)數(shù)據(jù)連接信息
 conn_info = input('請(qǐng)輸入mysql數(shù)據(jù)庫連接信息(格式為：主機(jī)IP,用戶名,登錄密碼,數(shù)據(jù)庫名)，逗號(hào)分隔且輸入順序不能亂，例如：192.168.0.1,root,root,test_db：')
 conn_list = conn_info.split(',')
 while conn_info == '' or len(conn_list) != 4:
 conn_info = input('請(qǐng)正確輸入mysql數(shù)據(jù)庫連接信息(格式為：主機(jī)IP,用戶名,登錄密碼,數(shù)據(jù)庫名)，逗號(hào)分隔且輸入順序不能亂，例如：192.168.0.1,root,root,test_db：')
 conn_list = conn_info.split(',')
 # 測試數(shù)據(jù)庫連接問題
 dd_test = DataDict(conn_list)
 db_conn = dd_test.test_conn(conn_list)
 while db_conn == False:
 conn_info = input('請(qǐng)正確輸入mysql數(shù)據(jù)庫連接信息(格式為：主機(jī)IP,用戶名,登錄密碼,數(shù)據(jù)庫名)，逗號(hào)分隔且輸入順序不能亂，例如：192.168.0.1,root,root,test_db：')
 conn_list = conn_info.split(',')
 if len(conn_list) != 4:
  continue
 dd_test = DataDict(conn_list)
 db_conn = dd_test.test_conn(conn_list)
 # 輸入數(shù)據(jù)表名稱
 table_s = input('請(qǐng)輸入數(shù)據(jù)庫表名(例如：t_order)，如需輸入多個(gè)表名請(qǐng)用英文逗號(hào)分隔(例如：t_order,t_goods)，結(jié)束使用請(qǐng)輸入q：')
 dd = DataDict(conn_list)
 while table_s != 'q':
 dd.run(table_s)
 table_s = input('繼續(xù)使用請(qǐng)輸入數(shù)據(jù)庫表名（例如t_order），如需輸入多個(gè)表名請(qǐng)用英文逗號(hào)分隔（例如t_order,t_goods），結(jié)束使用請(qǐng)輸入q）：')
 else:
 print('謝謝使用，再見……')
 time.sleep(1)