AI驱动的数据分析：从自然语言查询到自动洞察

class TextToSQLEngine:
    def __init__(self, llm, schema_extractor):
        self.llm = llm
        self.schema = schema_extractor
    
    def query(self, natural_language, database):
        """自然语言转SQL"""
        # 1. 提取数据库schema
        schema_info = self.schema.extract(database)
        # 包含：表结构、字段说明、外键关系、示例数据
        
        # 2. 意图理解
        intent = self._understand_intent(natural_language)
        # 聚合/过滤/排序/连接/窗口函数
        
        # 3. 生成SQL
        sql = self.llm.generate(f"""
        数据库Schema：
        {schema_info}
        
        用户问题：{natural_language}
        意图分析：{intent}
        
        生成PostgreSQL查询。要求：
        1. 只使用SELECT语句
        2. 使用表别名提高可读性
        3. 添加LIMIT防止全表扫描
        4. 处理NULL值
        5. 使用COALESCE处理空值
        """)
        
        # 4. SQL验证和优化
        validated_sql = self._validate_and_optimize(sql, database)
        
        return validated_sql

class SchemaExtractor:
    def extract(self, database):
        """提取数据库schema信息"""
        schema = {
            "tables": {},
            "relationships": [],
            "sample_data": {},
            "statistics": {}
        }
        
        for table in database.get_tables():
            schema["tables"][table.name] = {
                "columns": [],
                "row_count": table.row_count(),
                "description": table.comment or ""
            }
            
            for col in table.columns:
                schema["tables"][table.name]["columns"].append({
                    "name": col.name,
                    "type": col.type,
                    "nullable": col.nullable,
                    "description": col.comment or "",
                    "sample_values": table.sample_values(col.name, n=5)
                })
        
        # 外键关系
        schema["relationships"] = database.get_foreign_keys()
        
        return schema

class MultiTurnQueryOptimizer:
    def __init__(self):
        self.query_history = []
    
    def refine_query(self, current_question, previous_results):
        """基于历史查询优化当前查询"""
        context = ""
        for q, r in self.query_history[-3:]:  # 最近3轮
            context += f"之前问过：{q}\n结果摘要：{r}\n\n"
        
        refined = self.llm.generate(f"""
        之前的对话历史：
        {context}
        
        当前问题：{current_question}
        
        如果当前问题与之前相关，生成增量查询。
        如果是全新问题，生成独立查询。
        如果需要对比，引用之前的结果。
        """)
        
        return refined

class InsightDetector:
    def __init__(self, llm):
        self.llm = llm
    
    def detect_insights(self, data, dimensions, metrics):
        """自动发现数据中的洞察"""
        insights = []
        
        # 1. 趋势检测
        trends = self._detect_trends(data, dimensions["time"], metrics)
        insights.extend(trends)
        
        # 2. 异常检测
        anomalies = self._detect_anomalies(data, metrics)
        insights.extend(anomalies)
        
        # 3. 相关性发现
        correlations = self._find_correlations(data, metrics)
        insights.extend(correlations)
        
        # 4. 分群发现
        segments = self._discover_segments(data, dimensions)
        insights.extend(segments)
        
        # 5. AI解读
        for insight in insights:
            insight["explanation"] = self._explain(insight)
        
        return insights
    
    def _detect_anomalies(self, data, metrics):
        """统计异常检测"""
        anomalies = []
        for metric in metrics:
            values = data[metric]
            mean, std = values.mean(), values.std()
            
            # Z-score异常
            z_scores = (values - mean) / std
            outliers = data[abs(z_scores) > 3]
            
            if len(outliers) > 0:
                anomalies.append({
                    "type": "anomaly",
                    "metric": metric,
                    "severity": "high" if any(abs(z) > 5) else "medium",
                    "details": f"发现{len(outliers)}个异常点",
                    "data": outliers
                })
        
        return anomalies
    
    def _explain(self, insight):
        """AI解释洞察"""
        return self.llm.generate(f"""
        用简洁的语言解释以下数据发现：
        
        发现类型：{insight['type']}
        涉及指标：{insight.get('metric', 'N/A')}
        详情：{insight['details']}
        数据：{insight.get('data', 'N/A')}
        
        解释要求：
        1. 一句话说清楚发现了什么
        2. 可能的原因分析
        3. 建议的进一步分析方向
        """)

class AutoVisualizer:
    def visualize(self, data, question):
        """根据问题和数据自动选择最佳可视化方式"""
        chart_type = self.llm.generate(f"""
        用户问题：{question}
        数据特征：
        - 列：{list(data.columns)}
        - 行数：{len(data)}
        - 数据类型：{data.dtypes.to_dict()}
        
        选择最适合的可视化类型：
        1. 折线图（时间趋势）
        2. 柱状图（分类比较）
        3. 散点图（相关性）
        4. 饼图（占比）
        5. 热力图（多维交叉）
        6. 箱线图（分布）
        
        返回JSON：{{"chart_type": "...", "x": "...", "y": "...", "color": "..."}}
        """)
        
        config = json.loads(chart_type)
        return self._render(data, config)

class DataStoryteller:
    def narrate(self, data, insights, audience="executive"):
        """将数据分析结果转化为叙事"""
        story = self.llm.generate(f"""
        基于以下数据发现，写一份数据分析报告。
        
        目标受众：{audience}
        
        关键发现：
        {json.dumps(insights, ensure_ascii=False, indent=2)}
        
        数据摘要：
        {data.describe().to_string()}
        
        报告结构：
        1. 执行摘要（3句话概括最重要的发现）
        2. 详细分析（每个发现的深入解读）
        3. 异常与风险（需要关注的问题）
        4. 机会与建议（可执行的行动建议）
        5. 下一步分析方向
        
        语言要求：
        - {audience}级别的语言（避免/使用技术术语）
        - 用数据说话（引用具体数字）
        - 结论先行（每个段落先给结论再给依据）
        """)
        
        return story

class EnterpriseDataAgent:
    def __init__(self, llm, database, data_warehouse):
        self.llm = llm
        self.db = database
        self.dw = data_warehouse
        self.sql_engine = TextToSQLEngine(llm, SchemaExtractor())
        self.insight_detector = InsightDetector(llm)
        self.visualizer = AutoVisualizer()
        self.storyteller = DataStoryteller()
    
    def analyze(self, question):
        """端到端数据分析"""
        # 1. 理解问题
        analysis_plan = self._plan_analysis(question)
        
        # 2. 数据获取
        data = self._fetch_data(analysis_plan)
        
        # 3. 自动分析
        insights = self.insight_detector.detect_insights(
            data, 
            analysis_plan["dimensions"],
            analysis_plan["metrics"]
        )
        
        # 4. 可视化
        charts = [self.visualizer.visualize(data, question)]
        
        # 5. 叙事
        narrative = self.storyteller.narrate(data, insights)
        
        return {
            "question": question,
            "sql": analysis_plan["sql"],
            "data": data,
            "insights": insights,
            "charts": charts,
            "report": narrative
        }

class SQLSafetyGuard:
    def check(self, sql):
        """SQL安全检查"""
        forbidden = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE"]
        sql_upper = sql.upper()
        
        for keyword in forbidden:
            if keyword in sql_upper:
                return {"safe": False, "reason": f"禁止{keyword}操作"}
        
        # 检查是否有LIMIT
        if "LIMIT" not in sql_upper:
            sql = sql.rstrip(';') + " LIMIT 1000;"
        
        # 行数限制
        if "LIMIT" in sql_upper:
            limit_value = extract_limit(sql)
            if limit_value > 10000:
                return {"safe": False, "reason": "结果集过大"}
        
        return {"safe": True, "sql": sql}

class DataAccessController:
    def check_access(self, user, sql, tables):
        """检查用户对涉及表的数据权限"""
        for table in tables:
            if not self.has_permission(user, table):
                return False, f"无权限访问表: {table}"
        
        # 敏感数据检查
        sensitive_columns = self._find_sensitive(sql)
        for col in sensitive_columns:
            if not self.has_column_permission(user, col):
                return False, f"无权限访问敏感字段: {col}"
        
        return True, None