mirror of
https://github.com/csunny/DB-GPT.git
synced 2025-10-14 21:30:58 +00:00
136 lines
4.5 KiB
Python
136 lines
4.5 KiB
Python
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())
|