DB-GPT/pilot/summary/mysql_db_summary.py

import json

from pilot.configs.config import Config
from pilot.summary.db_summary import DBSummary, TableSummary, FieldSummary, IndexSummary

CFG = Config()


class MysqlSummary(DBSummary):
    """Get mysql summary template."""

    def __init__(self, name):
        self.name = name
        self.type = "MYSQL"
        self.summery = (
            """database name:{name}, database type:{type}, table infos:{table_info}"""
        )
        self.tables = {}
        self.tables_info = []
        self.vector_tables_info = []
        # self.tables_summary = {}

        self.db = CFG.local_db
        self.db.get_session(name)
        self.metadata = """user info :{users}, grant info:{grant}, charset:{charset}, collation:{collation}""".format(
            users=self.db.get_users(),
            grant=self.db.get_grants(),
            charset=self.db.get_charset(),
            collation=self.db.get_collation(),
        )
        tables = self.db.get_table_names()
        self.table_comments = self.db.get_table_comments(name)
        for table_comment in self.table_comments:
            self.tables_info.append(
                "table name:{table_name},table description:{table_comment}".format(
                    table_name=table_comment[0], table_comment=table_comment[1]
                )
            )
            vector_table = json.dumps(
                {"table_name": table_comment[0], "table_description": table_comment[1]}
            )
            self.vector_tables_info.append(
                vector_table.encode("utf-8").decode("unicode_escape")
            )

        for table_name in tables:
            table_summary = MysqlTableSummary(self.db, name, table_name)
            self.tables[table_name] = table_summary.get_summery()
            # self.tables_info.append(table_summary.get_summery())

    def get_summery(self):
        if CFG.SUMMARY_CONFIG == "VECTOR":
            return self.vector_tables_info
        else:
            return self.summery.format(
                name=self.name, type=self.type, table_info=";".join(self.tables_info)
            )

    def get_table_summary(self):
        return self.tables

    def get_table_comments(self):
        return self.table_comments


class MysqlTableSummary(TableSummary):
    """Get mysql table summary template."""

    def __init__(self, instance, dbname, name):
        self.name = name
        self.dbname = dbname
        self.summery = """database name:{dbname}, table name:{name}, have columns info: {fields}, have indexes info: {indexes}"""
        self.fields = []
        self.fields_info = []
        self.indexes = []
        self.indexes_info = []
        self.db = instance
        fields = self.db.get_fields(name)
        indexes = self.db.get_indexes(name)
        for field in fields:
            field_summary = MysqlFieldsSummary(field)
            self.fields.append(field_summary)
            self.fields_info.append(field_summary.get_summery())

        for index in indexes:
            index_summary = MysqlIndexSummary(index)
            self.indexes.append(index_summary)
            self.indexes_info.append(index_summary.get_summery())

    def get_summery(self):
        return self.summery.format(
            name=self.name,
            dbname=self.dbname,
            fields=";".join(self.fields_info),
            indexes=";".join(self.indexes_info),
        )


class MysqlFieldsSummary(FieldSummary):
    """Get mysql field summary template."""

    def __init__(self, field):
        self.name = field[0]
        self.summery = """column name:{name}, column data type:{data_type}, is nullable:{is_nullable}, default value is:{default_value}, comment is:{comment} """
        self.data_type = field[1]
        self.default_value = field[2]
        self.is_nullable = field[3]
        self.comment = field[4]

    def get_summery(self):
        return self.summery.format(
            name=self.name,
            data_type=self.data_type,
            is_nullable=self.is_nullable,
            default_value=self.default_value,
            comment=self.comment,
        )


class MysqlIndexSummary(IndexSummary):
    """Get mysql index summary template."""

    def __init__(self, index):
        self.name = index[0]
        self.summery = """index name:{name}, index bind columns:{bind_fields}"""
        self.bind_fields = index[1]

    def get_summery(self):
        return self.summery.format(name=self.name, bind_fields=self.bind_fields)


if __name__ == "__main__":
    summary = MysqlSummary("db_test")
    print(summary.get_summery())