def calculate_price(base_fee, improvement):  """根据性能提升比例动态定价"""  if improvement > 300:    # 提升300%+  return base_fee * 2.0  # 卓越性能溢价  elif improvement > 100:  # 提升100%-300%  return base_fee * 1.5  elif improvement > 30:   # 提升30%-100%  return base_fee  else:                    # 提升不足30%  return base_fee * 0.8  # 效果打折

# 创建项目目录  
mkdir sql-optimizer && cd sql-optimizer  
mkdir {php-app,python-ai,web-ui}  # 安装Python依赖  
cd python-ai  
python -m venv venv  
source venv/bin/activate  
pip install xgboost==1.7 sqlglot redis pandas scikit-learn  # 安装PHP依赖  
cd ../php-app  
composer require predis/predis illuminate/database guzzlehttp/guzzle

import sqlglot  
from sqlglot import expressions as exp  class SQLFeatureExtractor:  def extract(self, sql):  parsed = sqlglot.parse_one(sql)  features = {}  # 1. 基础特征  features['join_count'] = len(parsed.find_all(exp.Join))  features['subquery_depth'] = self._calc_subquery_depth(parsed)  features['where_condition_count'] = len(parsed.find_all(exp.Where))  # 2. 表特征（需连接数据库获取）  features['table_sizes'] = self._get_table_sizes(parsed)  # 3. 条件字段分析  where_conditions = parsed.find(exp.Where)  if where_conditions:  features['condition_columns'] = [  col.name for col in where_conditions.find_all(exp.Column)  ]  return features  def _calc_subquery_depth(self, node, depth=0):  """递归计算子查询深度"""  max_depth = depth  for subquery in node.find_all(exp.Subquery):  sub_depth = self._calc_subquery_depth(subquery, depth+1)  if sub_depth > max_depth:  max_depth = sub_depth  return max_depth  def _get_table_sizes(self, parsed):  """获取表大小（模拟实现）"""  # 生产环境需连接数据库获取真实大小  return {t.name: 1000000 for t in parsed.find_all(exp.Table)}

import xgboost as xgb  
import numpy as np  
from .feature_extractor import SQLFeatureExtractor  class SQLOptimizerModel:  def __init__(self, model_path='model.xgb'):  self.model = xgb.Booster(model_file=model_path)  self.feature_extractor = SQLFeatureExtractor()  self.feature_names = [  'join_count', 'subquery_depth', 'where_condition_count',  'table_size_score', 'condition_column_score'  ]  def predict(self, sql):  # 提取特征  raw_features = self.feature_extractor.extract(sql)  # 特征工程  features = self._engineer_features(raw_features)  # 模型预测  dmatrix = xgb.DMatrix(  [features],   feature_names=self.feature_names  )  prediction = self.model.predict(dmatrix)[0]  return self._decode_prediction(prediction, sql)  def _engineer_features(self, raw):  """将原始特征转换为模型输入"""  # 1. 表大小评分（最大表越小分越高）  max_table_size = max(raw['table_sizes'].values()) if raw['table_sizes'] else 0  table_score = 1 / (max_table_size / 1000000 + 0.01)  # 2. 条件字段评分（字段区分度）  # 生产环境需从数据库获取Cardinality  cond_score = len(raw.get('condition_columns', [])) * 0.2  return [  raw['join_count'],  raw['subquery_depth'],  raw['where_condition_count'],  table_score,  cond_score  ]  def _decode_prediction(self, pred, original_sql):  """将模型输出转为优化方案"""  # 示例逻辑：pred值>0.7推荐索引，否则重写SQL  if pred > 0.7:  return {  'type': 'ADD_INDEX',  'table': self._detect_main_table(original_sql),  'columns': self._suggest_index_columns(original_sql),  'confidence': float(pred)  }  else:  return {  'type': 'REWRITE_SQL',  'optimized_sql': self._rewrite_sql(original_sql),  'confidence': float(1 - pred)  }  def _detect_main_table(self, sql):  """识别主表（简化实现）"""  parsed = sqlglot.parse_one(sql)  return parsed.find(exp.Table).name  def _suggest_index_columns(self, sql):  """推荐索引字段（简化实现）"""  parsed = sqlglot.parse_one(sql)  where = parsed.find(exp.Where)  return [col.name for col in where.find_all(exp.Column)][:3]  def _rewrite_sql(self, sql):  """SQL重写（示例：转换IN为JOIN）"""  parsed = sqlglot.parse_one(sql)  # 查找IN子查询  for in_pred in parsed.find_all(exp.In):  if isinstance(in_pred.this, exp.Subquery):  # 转换为JOIN  new_join = exp.Join(  this=in_pred.this,  on=exp.EQ(  this=in_pred.expression,  expression=exp.Column(this=in_pred.args['this'])  )  )  in_pred.replace(new_join)  return parsed.sql()

import redis  
import json  
from .model_service import SQLOptimizerModel  r = redis.Redis(host='localhost', port=6379, db=0)  
model = SQLOptimizerModel()  def process_queue():  while True:  # 从队列获取任务  _, task_json = r.brpop('sql_optimize_tasks', timeout=30)  if not task_json:  continue  task = json.loads(task_json)  print(f"Processing task {task['id']}")  try:  # 执行优化分析  result = model.predict(task['sql'])  # 保存结果  r.set(f"optimize_result:{task['id']}", json.dumps({  'status': 'completed',  'result': result  }))  except Exception as e:  r.set(f"optimize_result:{task['id']}", json.dumps({  'status': 'failed',  'error': str(e)  }))  if __name__ == "__main__":  print("Starting SQL optimizer worker...")  process_queue()

<?php  
namespace App\Http\Middleware;  use Closure;  
use Illuminate\Support\Facades\DB;  
use Illuminate\Support\Facades\Redis;  class SlowQueryMonitor  
{  const SLOW_THRESHOLD = 0.1; // 100ms  public function handle($request, Closure $next)  {  // 开始监听  DB::enableQueryLog();  $start = microtime(true);  $response = $next($request);  // 检测慢查询  foreach (DB::getQueryLog() as $log) {  $duration = $log['time'] / 1000;  if ($duration > self::SLOW_THRESHOLD) {  $this->reportSlowQuery($log['query'], $duration);  }  }  return $response;  }  protected function reportSlowQuery($sql, $duration)  {  $taskId = uniqid('slow_');  // 发送到Redis队列  Redis::rpush('sql_optimize_tasks', json_encode([  'id' => $taskId,  'sql' => $sql,  'duration' => $duration,  'timestamp' => time()  ]));  // 记录任务ID用于后续追踪  Redis::setex("task:{$taskId}:status", 3600, 'pending');  }  
}

<?php  
namespace App\Services;  use Exception;  
use Illuminate\Support\Facades\DB;  
use Illuminate\Support\Facades\Log;  class SafeIndexDeployer  
{  public function execute(string $ddl, string $table)  {  if ($this->isPeakHour()) {  throw new Exception('禁止在业务高峰期执行DDL');  }  $backupTable = "backup_{$table}_" . time();  try {  // 1. 创建备份  DB::statement("CREATE TABLE {$backupTable} LIKE {$table}");  DB::statement("INSERT INTO {$backupTable} SELECT * FROM {$table}");  // 2. 执行DDL  DB::statement($ddl);  // 3. 验证基本功能  $this->validateTable($table);  return true;  } catch (Exception $e) {  // 4. 失败回滚  $this->rollback($table, $backupTable);  throw $e;  } finally {  // 5. 清理备份（成功时）  if (isset($e) === false) {  DB::statement("DROP TABLE {$backupTable}");  }  }  }  protected function isPeakHour(): bool  {  $hour = date('H');  return $hour >= 9 && $hour <= 18;  }  protected function validateTable($table)  {  // 简单验证：能否查询1条记录  DB::table($table)->first();  }  protected function rollback($originalTable, $backupTable)  {  DB::statement("DROP TABLE IF EXISTS {$originalTable}");  DB::statement("RENAME TABLE {$backupTable} TO {$originalTable}");  }  
}

<?php  namespace App\Console\Commands;  use Illuminate\Console\Command;  
use Illuminate\Support\Facades\Redis;  
use App\Services\SafeIndexDeployer;  class ProcessOptimization extends Command  
{  protected $signature = 'optimize:process';  protected $description = 'Process SQL optimization results';  public function handle()  {  while (true) {  // 检查已完成任务  $keys = Redis::keys('optimize_result:*');  foreach ($keys as $key) {  $result = json_decode(Redis::get($key), true);  if ($result['status'] === 'completed') {  $this->applyOptimization($result['result']);  Redis::del($key);  }  }  sleep(10); // 每10秒检查一次  }  }  protected function applyOptimization(array $result)  {  $deployer = new SafeIndexDeployer();  if ($result['type'] === 'ADD_INDEX') {  $ddl = sprintf(  "ALTER TABLE %s ADD INDEX %s (%s)",  $result['table'],  $this->generateIndexName($result),  implode(',', $result['columns'])  );  $deployer->execute($ddl, $result['table']);  } elseif ($result['type'] === 'REWRITE_SQL') {  $this->saveRewriteSuggestion($result);  }  }  protected function generateIndexName(array $result): string  {  return 'idx_' . implode('_', $result['columns']) . '_' . substr(md5(time()), 0, 6);  }  protected function saveRewriteSuggestion(array $result)  {  // 存储到数据库供DBA审核  DB::table('sql_rewrites')->insert([  'original_sql' => $result.get('original_sql', ''),  'optimized_sql' => $result['optimized_sql'],  'confidence' => $result['confidence'],  'created_at' => now()  ]);  }  
}

<template>  <div class="slow-query-monitor">  <h3>实时慢查询追踪</h3>  <div class="controls">  <button @click="startMonitoring" :disabled="isMonitoring">  启动监控  </button>  <button @click="stopMonitoring" :disabled="!isMonitoring">  停止监控  </button>  </div>  <table v-if="slowQueries.length">  <thead>  <tr>  <th>SQL</th>  <th>耗时(s)</th>  <th>状态</th>  <th>操作</th>  </tr>  </thead>  <tbody>  <tr v-for="(query, index) in slowQueries" :key="index">  <td class="sql-preview">{{ truncateSQL(query.sql) }}</td>  <td :class="{'critical': query.duration > 1}">  {{ query.duration.toFixed(3) }}  </td>  <td>  <span :class="statusClass(query.status)">  {{ query.status }}  </span>  </td>  <td>  <button  @click="optimizeQuery(query)"  v-if="query.status === 'pending'"  >  AI优化  </button>  <a  v-else-if="query.status === 'completed' && query.reportUrl"  :href="query.reportUrl"  target="_blank"  >  查看报告  </a>  </td>  </tr>  </tbody>  </table>  <div v-else class="empty-state">  <p>未检测到慢查询，系统运行良好！</p>  </div>  </div>  
</template>  <script>  
import axios from 'axios';  export default {  data() {  return {  isMonitoring: false,  slowQueries: [],  ws: null  };  },  methods: {  startMonitoring() {  this.ws = new WebSocket('ws://localhost:8000/ws/slow-queries');  this.ws.onmessage = (event) => {  const query = JSON.parse(event.data);  this.slowQueries.unshift({  ...query,  status: 'pending',  reportUrl: null  });  };  this.isMonitoring = true;  },  stopMonitoring() {  if (this.ws) {  this.ws.close();  }  this.isMonitoring = false;  },  async optimizeQuery(query) {  try {  const index = this.slowQueries.findIndex(q => q.id === query.id);  this.slowQueries[index].status = 'processing';  const response = await axios.post('/api/optimize', {  sql: query.sql  });  this.slowQueries[index] = {  ...query,  status: 'completed',  reportUrl: response.data.reportUrl,  recommendation: response.data.recommendation  };  } catch (error) {  console.error('优化失败:', error);  const index = this.slowQueries.findIndex(q => q.id === query.id);  this.slowQueries[index].status = 'failed';  }  },  truncateSQL(sql, length = 50) {  return sql.length > length ? sql.substring(0, length) + '...' : sql;  },  statusClass(status) {  return {  'status-badge': true,  'pending': status === 'pending',  'processing': status === 'processing',  'completed': status === 'completed',  'failed': status === 'failed'  };  }  },  beforeUnmount() {  if (this.ws) {  this.ws.close();  }  }  
};  
</script>  <style scoped>  
.status-badge { padding: 3px 8px; border-radius: 4px; color: white; }  
.status-badge.pending { background-color: #ff9800; }  
.status-badge.processing { background-color: #2196f3; }  
.status-badge.completed { background-color: #4caf50; }  
.status-badge.failed { background-color: #f44336; }  
.critical { color: #f44336; font-weight: bold; }  
.sql-preview { max-width: 300px; overflow: hidden; text-overflow: ellipsis; }  
</style>

<template>  <div class="optimization-report">  <h3>SQL优化报告</h3>  <div class="original-sql">  <h4>原始SQL</h4>  <pre>{{ report.original_sql }}</pre>  <p>执行耗时: <strong>{{ report.original_duration }}s</strong></p>  </div>  <div class="recommendation">  <h4>AI优化建议</h4>  <div v-if="report.type === 'ADD_INDEX'">  <p>建议添加索引:  <code>ALTER TABLE {{ report.table }} ADD INDEX {{ report.index_name }} ({{ report.columns.join(', ') }})</code>  </p>  <p>置信度: <strong>{{ (report.confidence * 100).toFixed(1) }}%</strong></p>  <button @click="deployIndex" class="deploy-btn">  安全部署  </button>  </div>  <div v-else-if="report.type === 'REWRITE_SQL'">  <p>建议改写SQL:</p>  <pre>{{ report.optimized_sql }}</pre>  <p>置信度: <strong>{{ (report.confidence * 100).toFixed(1) }}%</strong></p>  <button @click="testRewrite" class="test-btn">  沙箱测试  </button>  </div>  </div>  <div v-if="performanceData" class="performance-comparison">  <h4>性能预测</h4>  <bar-chart :data="performanceData" />  </div>  </div>  
</template>  <script>  
import axios from 'axios';  
import BarChart from './BarChart.vue';  export default {  components: { BarChart },  props: ['report'],  data() {  return {  performanceData: null  };  },  mounted() {  this.loadPerformancePrediction();  },  methods: {  async loadPerformancePrediction() {  const response = await axios.post('/api/predict-performance', {  report: this.report  });  this.performanceData = [  { label: '优化前', value: this.report.original_duration },  { label: '预测优化后', value: response.data.predicted_duration }  ];  },  async deployIndex() {  try {  await axios.post('/api/deploy-index', {  ddl: `ALTER TABLE ${this.report.table} ADD INDEX ${this.report.index_name} (${this.report.columns.join(', ')})`  });  this.$emit('deployed');  } catch (error) {  console.error('部署失败:', error);  }  },  async testRewrite() {  const response = await axios.post('/api/test-sql', {  original: this.report.original_sql,  optimized: this.report.optimized_sql  });  console.log('测试结果:', response.data);  this.$emit('test-completed', response.data);  }  }  
};  
</script>

# 启动Redis  
docker run -p 6379:6379 redis  # 启动Python AI工作器  
cd python-ai && python worker.py  # 启动PHP服务  
cd php-app && php artisan serve --port=8000  # 启动队列处理器  
php artisan optimize:process  # 启动Vue前端  
cd web-ui && npm run dev

# 使用curl模拟请求  
curl -X POST http://localhost:8000/api/query \  -H "Content-Type: application/json" \  -d '{"sql": "SELECT * FROM orders WHERE user_id = 100 AND status = 'pending'"}'

# 在Python端添加性能预测模型  
class PerformancePredictor:  def predict(self, original_time, recommendation):  if recommendation['type'] == 'ADD_INDEX':  # 索引优化通常提升10-100倍  return original_time * 0.1  elif recommendation['type'] == 'REWRITE_SQL':  # 查询重写通常提升2-10倍  return original_time * 0.3

// 在PHP执行器添加权限检查  
protected function checkPermissions($table)  
{  $user = auth()->user();  if (!$user->can('alter-table', $table)) {  throw new UnauthorizedException("无权限修改表结构");  }  
}

# Python端实现反馈学习  
def update_model(self, sql, actual_improvement):  # 获取原始特征  features = self.extract_features(sql)  # 根据实际改进更新标签  new_label = 1.0 if actual_improvement > 0.5 else 0.0  # 增量训练模型  self.model.partial_fit([features], [new_label])

apiVersion: apps/v1  
kind: Deployment  
metadata:  name: sql-optimizer  
spec:  replicas: 3  selector:  matchLabels:  app: optimizer  template:  metadata:  labels:  app: optimizer  spec:  containers:  - name: optimizer  image: your-registry/sql-optimizer:1.8  env:  - name: REDIS_HOST  value: "redis-cluster"  - name: MODEL_PATH  value: "/models/production.xgb"  resources:  limits:  cpu: "2"  memory: 4Gi  livenessProbe:  exec:  command: ["python", "healthcheck.py"]  initialDelaySeconds: 30  volumeMounts:  - name: model-volume  mountPath: /models  volumes:  - name: model-volume  persistentVolumeClaim:  claimName: model-pvc  ---  
# 定时优化任务  
apiVersion: batch/v1beta1  
kind: CronJob  
metadata:  name: nightly-optimization  
spec:  schedule: "0 3 * * *"  jobTemplate:  spec:  template:  spec:  containers:  - name: optimizer  image: your-registry/sql-optimizer:1.8  command: ["python", "run_optimization.py", "--full-scan"]  restartPolicy: OnFailure

# php.ini 关键配置  
[MySQL]  
mysql.allow_local_infile = On  
mysql.trace_mode = Off  # 生产环境关闭Trace  # my.cnf 优化建议  
[mysqld]  
innodb_buffer_pool_size = 80%内存  
innodb_flush_log_at_trx_commit = 2  # 兼顾性能与安全  
tmp_table_size = 64M  
max_heap_table_size = 64M