citu_vanna/schema_tools/sql_validator.py

"""
SQL验证器命令行入口
用于验证Question-SQL对中的SQL语句是否有效
"""

import argparse
import asyncio
import sys
import os
from pathlib import Path

from schema_tools.sql_validation_agent import SQLValidationAgent
from schema_tools.utils.logger import setup_logging


def setup_argument_parser():
    """设置命令行参数解析器"""
    parser = argparse.ArgumentParser(
        description='SQL Validator - 验证Question-SQL对中的SQL语句',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例用法:
  # 基本使用（仅验证，不修改文件）
  python -m schema_tools.sql_validator --db-connection "postgresql://user:pass@localhost:5432/dbname" --input-file ./data.json
  
  # 仅删除无效SQL，不进行LLM修复
  python -m schema_tools.sql_validator --db-connection "postgresql://user:pass@localhost:5432/dbname" --input-file ./data.json --modify-original-file
  
  # 启用LLM修复功能（需要同时指定文件修改参数）
  python -m schema_tools.sql_validator --db-connection "postgresql://user:pass@localhost:5432/dbname" --input-file ./data.json --enable-llm-repair --modify-original-file
  
  # 指定输出目录
  python -m schema_tools.sql_validator --db-connection "postgresql://user:pass@localhost:5432/dbname" --input-file ./data.json --output-dir ./reports
  
  # 启用详细日志
  python -m schema_tools.sql_validator --db-connection "postgresql://user:pass@localhost:5432/dbname" --input-file ./data.json --verbose
        """
    )
    
    # 必需参数
    parser.add_argument(
        '--db-connection',
        required=True,
        help='数据库连接字符串 (postgresql://user:pass@host:port/dbname)'
    )
    
    parser.add_argument(
        '--input-file',
        required=True,
        help='输入的JSON文件路径（包含Question-SQL对）'
    )
    
    # 可选参数
    parser.add_argument(
        '--output-dir',
        help='验证报告输出目录（默认为输入文件同目录）'
    )
    
    parser.add_argument(
        '--max-concurrent',
        type=int,
        help='最大并发验证数（覆盖配置文件设置）'
    )
    
    parser.add_argument(
        '--batch-size',
        type=int,
        help='批处理大小（覆盖配置文件设置）'
    )
    
    parser.add_argument(
        '--timeout',
        type=int,
        help='单个SQL验证超时时间（秒）'
    )
    
    parser.add_argument(
        '--verbose', '-v',
        action='store_true',
        help='启用详细日志输出'
    )
    
    parser.add_argument(
        '--log-file',
        help='日志文件路径'
    )
    
    parser.add_argument(
        '--dry-run',
        action='store_true',
        help='仅读取和解析文件，不执行验证'
    )
    
    parser.add_argument(
        '--save-json',
        action='store_true',
        help='同时保存详细的JSON报告'
    )
    
    parser.add_argument(
        '--enable-llm-repair',
        action='store_true',
        help='启用LLM自动修复功能'
    )
    
    parser.add_argument(
        '--modify-original-file',
        action='store_true',
        help='修改原始JSON文件（删除无效SQL，如果启用LLM修复则同时更新修复后的SQL）'
    )
    
    return parser


def apply_config_overrides(args):
    """应用命令行参数覆盖配置"""
    from schema_tools.config import SCHEMA_TOOLS_CONFIG
    
    sql_config = SCHEMA_TOOLS_CONFIG['sql_validation']
    
    if args.max_concurrent:
        sql_config['max_concurrent_validations'] = args.max_concurrent
        print(f"覆盖并发数配置: {args.max_concurrent}")
    
    if args.batch_size:
        sql_config['batch_size'] = args.batch_size
        print(f"覆盖批处理大小: {args.batch_size}")
    
    if args.timeout:
        sql_config['validation_timeout'] = args.timeout
        print(f"覆盖超时配置: {args.timeout}秒")
    
    if args.save_json:
        sql_config['save_detailed_json_report'] = True
        print(f"启用详细JSON报告保存")
    
    if args.enable_llm_repair:
        sql_config['enable_sql_repair'] = True
        print(f"启用LLM自动修复功能")
    else:
        sql_config['enable_sql_repair'] = False
        print(f"LLM修复功能已禁用")
    
    if args.modify_original_file:
        sql_config['modify_original_file'] = True
        print(f"启用原文件修改功能")
    else:
        sql_config['modify_original_file'] = False
        print(f"不修改原文件")


async def main():
    """主入口函数"""
    parser = setup_argument_parser()
    args = parser.parse_args()
    
    # 设置日志
    setup_logging(
        verbose=args.verbose,
        log_file=args.log_file,
        log_dir=os.path.join(args.output_dir, 'logs') if args.output_dir else None
    )
    
    # 验证参数
    if not os.path.exists(args.input_file):
        print(f"错误: 输入文件不存在: {args.input_file}")
        sys.exit(1)
    
    input_path = Path(args.input_file)
    if not input_path.suffix.lower() == '.json':
        print(f"警告: 输入文件可能不是JSON格式: {args.input_file}")
    
    # 应用配置覆盖
    apply_config_overrides(args)
    
    try:
        # 创建SQL验证Agent
        agent = SQLValidationAgent(
            db_connection=args.db_connection,
            input_file=args.input_file,
            output_dir=args.output_dir
        )
        
        # 显示运行信息
        print(f"🚀 开始SQL验证...")
        print(f"📁 输入文件: {args.input_file}")
        if args.output_dir:
            print(f"📁 输出目录: {args.output_dir}")
        print(f"🔗 数据库: {_mask_db_connection(args.db_connection)}")
        
        if args.dry_run:
            print("\n🔍 执行预检查模式...")
            # 仅读取和验证文件格式
            questions_sqls = await agent._load_questions_sqls()
            print(f"✅ 成功读取 {len(questions_sqls)} 个Question-SQL对")
            print("📊 SQL样例:")
            for i, qs in enumerate(questions_sqls[:3], 1):
                print(f"  {i}. {qs['question']}")
                print(f"     SQL: {qs['sql'][:100]}{'...' if len(qs['sql']) > 100 else ''}")
                print()
            sys.exit(0)
        
        # 执行验证
        report = await agent.validate()
        
        # 输出结果
        success_rate = report['summary']['success_rate']
        
        if success_rate >= 0.9:  # 90%以上成功率
            print(f"\n🎉 验证完成，成功率: {success_rate:.1%}")
            exit_code = 0
        elif success_rate >= 0.7:  # 70%-90%成功率
            print(f"\n⚠️  验证完成，成功率较低: {success_rate:.1%}")
            exit_code = 1
        else:  # 70%以下成功率
            print(f"\n❌ 验证完成，成功率过低: {success_rate:.1%}")
            exit_code = 2
        
        print(f"📊 详细结果: {report['summary']['valid_sqls']}/{report['summary']['total_questions']} SQL有效")
        
        sys.exit(exit_code)
        
    except KeyboardInterrupt:
        print("\n\n⏹️  用户中断，程序退出")
        sys.exit(130)
    except Exception as e:
        print(f"\n❌ 程序执行失败: {e}")
        if args.verbose:
            import traceback
            traceback.print_exc()
        sys.exit(1)


def _mask_db_connection(conn_str: str) -> str:
    """隐藏数据库连接字符串中的敏感信息"""
    import re
    return re.sub(r'://[^:]+:[^@]+@', '://***:***@', conn_str)


if __name__ == "__main__":
    asyncio.run(main())