GBase8a数据库设计文档导出工具

1   主要解决问题

（1）考虑到有的项目，客户有需要导出数据库内全部表对象的设计文档的要求，特开发了此工具。
（2）设计文档可以按照WORD格式个PDF格式导出。 
（3）设计文档按照vc、库、表、列的顺序输出，包含部分注释。

2   部署方式

（1） 修改脚本中以下片段为使用的环境参数：
       connection = pymysql.connect(
           host='192.168.1.5',
           user='gbase',
           password='gbase20110531',
           database='gbase',
           port=5258
       )

3   使用方式

（1）直接运行python db_doc.py（输出word格式） 或者python db_pdf.py（输出pdf格式）

（2）在脚本同目录下，会生成database_design_document.pdf或者database_design_document.docx文件

（3）注：该工具基于python3.8开发

4   演示样例

（1） 输出的PDF文件格式

(2) 输出WORD文件格式

5   参考文件

（1）db_doc.py文件

import pymysql
from docx import Document
from docx.shared import Pt


def get_db_connection():
    try:
        connection = pymysql.connect(
            host='192.168.1.5',
            user='gbase',
            password='gbase20110531',
            database='gbase',
            port=5258
        )
        return connection
    except pymysql.Error as e:
        print(f"数据库连接错误: {e}")
        return None


def get_all_dbs_tables(connection):
    try:
        cursor = connection.cursor()
        query = """
            SELECT dbName, tbName, vc_id 
            FROM table_distribution 
            WHERE dbName NOT IN ('information_schema', 'performance_schema', 'gbase', 'gclusterdb')
        """
        cursor.execute(query)
        return cursor.fetchall()
    except pymysql.Error as e:
        print(f"查询所有数据库和表时出错: {e}")
        return []


def get_all_vc_info(connection):
    try:
        cursor = connection.cursor()
        query = "SELECT ID, NAME FROM information_schema.vc"
        cursor.execute(query)
        return cursor.fetchall()
    except pymysql.Error as e:
        print(f"查询所有虚拟集群信息时出错: {e}")
        return []


def get_table_comment(connection, vc_id, db_name, table_name):
    try:
        cursor = connection.cursor()
        query = """
            SELECT TABLE_COMMENT 
            FROM information_schema.TABLES 
            WHERE VC_ID = %s AND TABLE_SCHEMA = %s AND TABLE_NAME = %s
        """
        cursor.execute(query, (vc_id, db_name, table_name))
        result = cursor.fetchone()
        return result[0] if result and result[0] else "注释：无"
    except pymysql.Error as e:
        print(f"查询表注释时出错: {e}")
        return "注释：无"


def get_table_columns(connection, vc_name, db_name, table_name):
    try:
        cursor = connection.cursor()
        query = """
            SELECT COLUMN_NAME, COLUMN_TYPE, COLUMN_COMMENT, ORDINAL_POSITION 
            FROM information_schema.COLUMNS 
            WHERE TABLE_VC = %s AND TABLE_SCHEMA = %s AND TABLE_NAME = %s
            ORDER BY ORDINAL_POSITION
        """
        cursor.execute(query, (vc_name, db_name, table_name))
        columns = cursor.fetchall()
        formatted_columns = []
        for column_name, column_type, column_comment, ordinal_position in columns:
            column_comment = column_comment if column_comment else "-"
            formatted_columns.append((column_name, column_type, column_comment, ordinal_position))
        return formatted_columns
    except pymysql.Error as e:
        print(f"查询表列信息时出错: {e}")
        return []


def generate_doc():
    connection = get_db_connection()
    if not connection:
        return

    all_dbs_tables = get_all_dbs_tables(connection)
    all_vc_info = get_all_vc_info(connection)
    vc_dict = {vc_id: vc_name for vc_id, vc_name in all_vc_info}

    doc = Document()
    doc.add_heading('GBase8a数据库设计文档', 0)

    vc_index = 1
    for vc_id, vc_name in sorted(vc_dict.items()):
        vc_heading = doc.add_heading(f'{vc_index}. 虚拟集群名: {vc_name}', 1)
        vc_heading.runs[0].font.size = Pt(14)

        db_dict = {}
        for db_name, table_name, table_vc_id in all_dbs_tables:
            if table_vc_id == vc_id:
                if db_name not in db_dict:
                    db_dict[db_name] = []
                db_dict[db_name].append(table_name)

        db_index = 1
        for db_name, table_names in sorted(db_dict.items()):
            db_heading = doc.add_heading(f'{vc_index}.{db_index} 库名: {db_name}', 2)
            db_heading.runs[0].font.size = Pt(12)

            table_index = 1
            for table_name in table_names:
                table_comment = get_table_comment(connection, vc_id, db_name, table_name)
                table_heading = doc.add_heading(f'{vc_index}.{db_index}.{table_index} 表名: {table_name} ({table_comment})', 3)
                table_heading.runs[0].font.size = Pt(10)

                columns = get_table_columns(connection, vc_name, db_name, table_name)
                table = doc.add_table(rows=1, cols=3)
                hdr_cells = table.rows[0].cells
                hdr_cells[0].text = '列名'
                hdr_cells[1].text = '列类型'
                hdr_cells[2].text = '列注释'

                for column_name, column_type, column_comment, _ in columns:
                    row_cells = table.add_row().cells
                    row_cells[0].text = column_name
                    row_cells[1].text = column_type
                    row_cells[2].text = column_comment

                table_index += 1
            db_index += 1
        vc_index += 1

    connection.close()
    doc.save('database_design_document.docx')
    print("数据库设计文档已生成: database_design_document.docx")


if __name__ == "__main__":
    generate_doc()
    

（2）db_pdf.py文件

import pymysql
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.platypus import Table, TableStyle
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont

 
pdfmetrics.registerFont(TTFont('SimSun', 'simsun.ttc'))


def get_db_connection():
    try:
        connection = pymysql.connect(
            host='192.168.1.5',
            user='gbase',
            password='gbase20110531',
            database='gbase',
            port=5258
        )
        return connection
    except pymysql.Error as e:
        print(f"数据库连接错误: {e}")
        return None


def get_all_dbs_tables(connection):
    try:
        cursor = connection.cursor()
        query = """
            SELECT dbName, tbName, vc_id 
            FROM table_distribution 
            WHERE dbName NOT IN ('information_schema', 'performance_schema', 'gbase', 'gclusterdb')
        """
        cursor.execute(query)
        return cursor.fetchall()
    except pymysql.Error as e:
        print(f"查询所有数据库和表时出错: {e}")
        return []


def get_all_vc_info(connection):
    try:
        cursor = connection.cursor()
        query = "SELECT ID, NAME FROM information_schema.vc"
        cursor.execute(query)
        return cursor.fetchall()
    except pymysql.Error as e:
        print(f"查询所有虚拟集群信息时出错: {e}")
        return []


def get_table_comment(connection, vc_id, db_name, table_name):
    try:
        cursor = connection.cursor()
        query = """
            SELECT TABLE_COMMENT 
            FROM information_schema.TABLES 
            WHERE VC_ID = %s AND TABLE_SCHEMA = %s AND TABLE_NAME = %s
        """
        cursor.execute(query, (vc_id, db_name, table_name))
        result = cursor.fetchone()
        return result[0] if result and result[0] else "注释：无"
    except pymysql.Error as e:
        print(f"查询表注释时出错: {e}")
        return "注释：无"


def get_table_columns(connection, vc_name, db_name, table_name):
    try:
        cursor = connection.cursor()
        query = """
            SELECT COLUMN_NAME, COLUMN_TYPE, COLUMN_COMMENT, ORDINAL_POSITION 
            FROM information_schema.COLUMNS 
            WHERE TABLE_VC = %s AND TABLE_SCHEMA = %s AND TABLE_NAME = %s
            ORDER BY ORDINAL_POSITION
        """
        cursor.execute(query, (vc_name, db_name, table_name))
        columns = cursor.fetchall()
        formatted_columns = []
        for column_name, column_type, column_comment, ordinal_position in columns:
            column_comment = column_comment if column_comment else "-"
            formatted_columns.append((column_name, column_type, column_comment, ordinal_position))
        return formatted_columns
    except pymysql.Error as e:
        print(f"查询表列信息时出错: {e}")
        return []


def generate_pdf():
    connection = get_db_connection()
    if not connection:
        return

    all_dbs_tables = get_all_dbs_tables(connection)
    all_vc_info = get_all_vc_info(connection)
    vc_dict = {vc_id: vc_name for vc_id, vc_name in all_vc_info}

    pdf = canvas.Canvas('database_design_document.pdf', pagesize=letter)
    styleSheet = getSampleStyleSheet()
    h1 = styleSheet['Heading1']
    h2 = styleSheet['Heading2']
    h3 = styleSheet['Heading3']
    body = styleSheet['BodyText']

 
    pdf.setFont('SimSun', h1.fontSize)
    pdf.drawString(100, 750, 'GBase8a数据库设计文档')

    y_position = 700
    vc_index = 1
    for vc_id, vc_name in sorted(vc_dict.items()):
 
        pdf.setFont('SimSun', h2.fontSize)
        vc_title = f'{vc_index}. 虚拟集群名: {vc_name}'
        pdf.drawString(100, y_position, vc_title)
        y_position -= 20

        db_dict = {}
        for db_name, table_name, table_vc_id in all_dbs_tables:
            if table_vc_id == vc_id:
                if db_name not in db_dict:
                    db_dict[db_name] = []
                db_dict[db_name].append(table_name)

        db_index = 1
        for db_name, table_names in sorted(db_dict.items()):
 
            pdf.setFont('SimSun', h3.fontSize)
            db_title = f'{vc_index}.{db_index} 库名: {db_name}'
            pdf.drawString(120, y_position, db_title)
            y_position -= 20

            table_index = 1
            for table_name in table_names:
                table_comment = get_table_comment(connection, vc_id, db_name, table_name)
                table_title = f'{vc_index}.{db_index}.{table_index} 表名: {table_name} ({table_comment})'
 
                pdf.setFont('SimSun', body.fontSize)
                pdf.drawString(140, y_position, table_title)
                y_position -= 20

                columns = get_table_columns(connection, vc_name, db_name, table_name)
                data = [('列名', '列类型', '列注释')]
                for column_name, column_type, column_comment, _ in columns:
                    data.append((column_name, column_type, column_comment))

                table = Table(data)
                table.setStyle(TableStyle([
                    ('BACKGROUND', (0, 0), (-1, 0), (0.9, 0.9, 0.9)),
                    ('TEXTCOLOR', (0, 0), (-1, 0), (0, 0, 0)),
                    ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
                    ('FONTNAME', (0, 0), (-1, 0), 'SimSun'),  
                    ('BOTTOMPADDING', (0, 0), (-1, 0), 8),
                    ('GRID', (0, 0), (-1, -1), 1, (0.5, 0.5, 0.5)),
                ]))

                tw, th = table.wrapOn(pdf, letter[0] - 100, letter[1] - y_position)
                if y_position - th < 50:   
                    pdf.showPage()
                    y_position = letter[1] - 50
                    pdf.setFont('SimSun', body.fontSize)
                table.drawOn(pdf, 140, y_position - th)
                y_position -= th + 20

                table_index += 1
            db_index += 1
        vc_index += 1

    connection.close()
    pdf.save()
    print("数据库设计文档已生成: database_design_document.pdf")


if __name__ == "__main__":
    generate_pdf()