1.1版本-新增where条件表达式值判断

Flask_src/requirements.txt

@@ -0,0 +1,8 @@
+Flask
+PyMySQL
+sql-metadata
+tabulate
+rich
+PyYAML
+vanna==0.0.36
+sqlparse

Flask_src/sql_alias.py

@@ -0,0 +1,11 @@
+def has_table_alias(table_alias):
+    if isinstance(table_alias, dict):
+        table_alias = {k.lower(): v.lower() for k, v in table_alias.items()}
+        if 'join' in table_alias or 'on' in table_alias or 'where' in table_alias or 'group by' in table_alias or 'order by' in table_alias or 'limit' in table_alias or not table_alias:
+           return False # 没有别名
+        else:
+           return True #有别名
+    elif isinstance(table_alias, list):
+        return False # 没有别名
+    else:
+        pass

Flask_src/sql_count_value.py

@@ -0,0 +1,115 @@
+import pymysql
+from rich.progress import Progress, TimeElapsedColumn, TextColumn, BarColumn
+
+def count_column_value(table_name, field_name, mysql_settings, sample_size):
+    with pymysql.connect(**mysql_settings) as conn:
+        with conn.cursor() as cursor:
+            """
+            在这个查询中，使用了CASE语句来判断数据行数是否小于100000。如果数据行数小于100000，则使用
+            (SELECT COUNT(*) FROM {table_name}) / 2
+            作为阈值，即表的实际大小除以2；否则使用 {sample_size} / 2 作为阈值。
+            """
+
+            # 如果你的数据库是MySQL 8.0，那么推荐用 CTE（公共表达式）的形式
+            '''
+            sql = f"""
+            WITH subquery AS (
+                SELECT {field_name}
+                FROM {table_name}
+                LIMIT {sample_size}
+            )
+            SELECT COUNT(*) as count
+            FROM subquery
+            GROUP BY {field_name}
+            HAVING COUNT(*) >= 
+                CASE 
+                    WHEN (SELECT COUNT(*) FROM subquery) < {sample_size} THEN (SELECT COUNT(*) FROM {table_name}) / 2 
+                    ELSE {sample_size} / 2 
+                END;
+            """
+            '''
+
+            # 默认采用子查询兼容MySQL 5.7版本
+            sql = f"""
+            SELECT COUNT(*) as count
+            FROM (
+                SELECT {field_name}
+                FROM {table_name}
+                LIMIT {sample_size}
+            ) AS subquery
+            GROUP BY {field_name}
+            HAVING COUNT(*) >= CASE WHEN (SELECT COUNT(*) FROM {table_name} LIMIT {sample_size}) < {sample_size} 
+            THEN (SELECT COUNT(*) FROM {table_name}) / 2 ELSE {sample_size} / 2 END;
+            """
+
+            # print(sql)
+            with Progress(TextColumn("[progress.description]{task.description}", justify="right"), BarColumn(), TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), TimeElapsedColumn()) as progress:
+                task = progress.add_task(f"[cyan]Executing SQL query [bold magenta]{table_name}[/bold magenta] [cyan]where_clause [bold magenta]{field_name}[/bold magenta]...", total=1)
+                cursor.execute(sql)
+                progress.update(task, completed=1)
+
+            results = cursor.fetchall()
+
+    if results:
+        # 如果有超过半数的重复数据
+        return results
+    else:
+        return False
+
+
+def count_column_clause_value(table_name, field_name, where_clause_value, mysql_settings, sample_size):
+    with pymysql.connect(**mysql_settings) as conn:
+        with conn.cursor() as cursor:
+
+            """
+            在这个查询中，使用了CASE语句来判断数据行数是否小于100000。如果数据行数小于100000，则使用
+            (SELECT COUNT(*) FROM {table_name}) / 2
+            作为阈值，即表的实际大小除以2；否则使用 {sample_size} / 2 作为阈值。
+            """
+
+            # 如果你的数据库是MySQL 8.0，那么推荐用 CTE（公共表达式）的形式
+            '''
+            sql = f"""
+            WITH subquery AS (
+                SELECT {field_name}
+                FROM {table_name}
+                LIMIT {sample_size}
+            )
+            SELECT COUNT(*) as count
+            FROM subquery
+            GROUP BY {field_name}
+            HAVING COUNT(*) >= 
+                CASE 
+                    WHEN (SELECT COUNT(*) FROM subquery) < {sample_size} THEN (SELECT COUNT(*) FROM {table_name}) / 2 
+                    ELSE {sample_size} / 2 
+                END;
+            """
+            '''
+
+            # 默认采用子查询兼容MySQL 5.7版本
+            sql = f"""
+            SELECT COUNT(*) as count
+            FROM (
+                SELECT {field_name}
+                FROM {table_name}
+                WHERE {where_clause_value}
+                LIMIT {sample_size}
+            ) AS subquery
+            GROUP BY {field_name}
+            HAVING COUNT(*) >= CASE WHEN (SELECT COUNT(*) FROM {table_name} LIMIT {sample_size}) < {sample_size} 
+            THEN (SELECT COUNT(*) FROM {table_name}) / 2 ELSE {sample_size} / 2 END;
+            """
+
+            #print(sql)
+            with Progress(TextColumn("[progress.description]{task.description}", justify="right"), BarColumn(), TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), TimeElapsedColumn()) as progress:
+                task = progress.add_task(f"[cyan]Executing SQL query [bold magenta]{table_name}[/bold magenta] [cyan]where_clause [bold magenta]{where_clause_value}[/bold magenta]...", total=1)
+                cursor.execute(sql)
+                progress.update(task, completed=1)
+
+            results = cursor.fetchall()
+
+    if results:
+        # 如果有超过半数的重复数据
+        return results
+    else:
+        return False

Flask_src/sql_extra.py

@@ -0,0 +1,26 @@
+import re
+
+def check_percent_position(string):
+    string = string.lower()
+    pattern = r".*like\s+'%|.*like\s+concat\(+'%|.*regexp\s+"
+    matches = re.findall(pattern, string)
+    if matches:
+        like_pattern = r"like\s+(?:concat\(.*?\)|'%%'|\'.*?%(?:.*?)?\')"
+        like_match = re.search(like_pattern, string)
+        if like_match:
+            return True, like_match.group()
+        #return True
+    return False, None
+
+
+def extract_function_index(string):
+    #pattern = r'\b(\w+)\('
+    #pattern = r'\b(\w+)\(.*\).*[>=<!=<>]'
+    pattern = r'\b(\w+(\(.*\).*[>=<!=<>]))'
+    matches = re.findall(pattern, string)
+    #function_indexes = set(matches)
+    function_indexes = [match[0] for match in matches]
+    if function_indexes:
+        return ', '.join(function_indexes)
+    else:
+        return False

Flask_src/sql_format_class.py

@@ -0,0 +1,12 @@
+import sqlparse
+
+class SQLFormatter:
+    def format_sql(self, sql_query):
+        """
+        格式化 SQL 查询语句
+        """
+        formatted_sql = sqlparse.format(sql_query, reindent=True, keyword_case='upper')
+
+        return formatted_sql
+
+

Flask_src/sql_index.py

@@ -0,0 +1,106 @@
+import textwrap
+from tabulate import tabulate
+import pymysql
+
+def execute_index_query(mysql_settings, database, table_name, index_columns):
+    index_columns = index_columns
+    index_columns = index_columns.split(',')
+    updated_columns = [f"'{column.strip()}'" for column in index_columns]
+    final_columns = ', '.join(updated_columns)
+    sql = f"SELECT TABLE_NAME,INDEX_NAME,COLUMN_NAME,CARDINALITY FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = '{database}' AND TABLE_NAME = '{table_name}' AND COLUMN_NAME IN ({final_columns})"
+    #print(sql)
+    try:
+        conn = pymysql.connect(**mysql_settings)
+        cur = conn.cursor()
+        cur.execute(sql)
+        index_result = cur.fetchall()
+
+        if not index_result:
+            print(f"没有检测到 {table_name} 表 字段 {final_columns} 有索引。")
+
+        # 提取列名
+        e_column_names = [desc[0] for desc in cur.description]
+
+        # 提取结果值并进行自动换行处理
+        e_result_values = []
+        for row in index_result:
+            values = list(row.values())
+            wrapped_values = [textwrap.fill(str(value), width=30) for value in values]
+            e_result_values.append(wrapped_values)
+
+        # 将结果格式化为表格（包含竖线）
+        e_table = tabulate(e_result_values, headers=e_column_names, tablefmt="grid", numalign="left")
+
+        return e_table
+
+    except pymysql.err.ProgrammingError as e:
+        print("MySQL 内部错误：",e)
+        return None
+    except Exception as e:
+        print("MySQL 内部错误：",e)
+        return None
+    finally:
+        if cur:
+            cur.close()
+        if conn:
+            conn.close()
+
+#########################################################
+
+def check_index_exist(mysql_settings, table_name, index_column):
+    show_index_sql = f"show index from {table_name} where Column_name = '{index_column}'"
+    try:
+        conn = pymysql.connect(**mysql_settings)
+        cur = conn.cursor()
+        cur.execute(show_index_sql)
+        index_result = cur.fetchall()
+
+        #if not index_result:
+            #print(f"没有检测到 {table_name} 表 字段 {final_columns} 有索引。")
+
+        return index_result
+
+    except pymysql.err.ProgrammingError as e:
+        print("MySQL 内部错误：",e)
+        return None
+    except Exception as e:
+        print("MySQL 内部错误：",e)
+        return None
+    finally:
+        if cur:
+            cur.close()
+        if conn:
+            conn.close()
+
+#########################################################
+
+def check_index_exist_multi(mysql_settings, database, table_name, index_columns, index_number):
+    index_columns = index_columns
+    index_columns = index_columns.split(',')
+    updated_columns = [f"'{column.strip()}'" for column in index_columns]
+    final_columns = ', '.join(updated_columns)
+    sql = f"SELECT TABLE_NAME,INDEX_NAME,COLUMN_NAME,CARDINALITY FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = '{database}' AND TABLE_NAME = '{table_name}' AND COLUMN_NAME IN ({final_columns}) GROUP BY INDEX_NAME HAVING COUNT(INDEX_NAME) = {index_number}"
+    #print(sql)
+    try:
+        conn = pymysql.connect(**mysql_settings)
+        cur = conn.cursor()
+        cur.execute(sql)
+        index_result = cur.fetchall()
+
+        if not index_result:
+            return None
+
+        return index_result
+
+    except pymysql.err.ProgrammingError as e:
+        print("MySQL 内部错误：",e)
+        return None
+    except Exception as e:
+        print("MySQL 内部错误：",e)
+        return None
+    finally:
+        if cur:
+            cur.close()
+        if conn:
+            conn.close()
+

Flask_src/sqlai.py

@@ -0,0 +1,36 @@
+from vanna.remote import VannaDefault
+
+def optimize_sql(original_sql):
+    """
+    调用 vanna.ai LLM 接口优化 SQL 查询语句，并返回优化后的 SQL 字符串。
+
+    Args:
+        original_sql (str): 原始的 SQL 查询语句。
+
+    Returns:
+        str: 优化后的 SQL 查询语句。 如果调用过程中发生错误，则返回包含错误信息的字符串。
+    """
+    try:
+        # 创建 VannaDefault 实例
+        vn = VannaDefault(model='sql_helper', api_key='xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx')
+
+        # 输出调用信息 (可选，仅用于调试)
+        print('\033[94m以下是调用的vanna.ai LLM接口.\033[0m')
+        print('优化前的SQL是：')
+        print(original_sql)
+        print('-' * 55)
+
+        # 获取优化后的 SQL
+        #vn.ask('How to optimize this SQL : {}'.format(original_sql))
+        optimized_sql = vn.generate_sql('How to optimize this SQL : {}'.format(original_sql))
+
+        # 输出优化后的 SQL (可选，仅用于调试)
+        print('\033[92m优化后的SQL是：\033[0m')
+
+        return optimized_sql
+
+    except Exception as e:
+        # 捕获异常，返回包含错误信息的字符串
+        error_message = f"调用 vanna.ai 优化出错: {e}"
+        print(error_message)  # 打印错误信息到控制台 (可选)
+        return error_message