测试、安全、数据从业者，手上有哪些好用的AI落地方案？

from openai import OpenAI
import json

class TestCaseGenerator:
    """基于LLM的测试用例自动生成器"""
    
    def __init__(self, base_url="http://localhost:8000/v1"):
        self.client = OpenAI(base_url=base_url, api_key="empty")
    
    def generate(self, requirement: str, api_spec: str = ""):
        """根据需求文档生成结构化测试用例"""
        
        system_prompt = """你是一位有10年经验的资深测试工程师，精通边界值分析、等价类划分、场景法等测试设计方法。

请根据给定的需求文档生成测试用例，要求：
1. 覆盖正常流程、异常流程、边界值
2. 每条用例包含：用例编号、模块、标题、前置条件、操作步骤、预期结果、优先级
3. 优先级：P0(冒烟必须通过)、P1(核心功能)、P2(边界和异常)、P3(次要场景)
4. 以JSON数组格式输出，方便程序解析"""

        user_prompt = f"""需求文档：
{requirement}

{'接口规格说明：' + chr(10) + api_spec if api_spec else ''}

请生成完整的测试用例列表。"""

        response = self.client.chat.completions.create(
            model="Qwen/Qwen2.5-7B-Instruct",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            temperature=0.3,
            max_tokens=4096,
            response_format={"type": "json_object"},
        )
        
        result = response.choices[0].message.content
        try:
            test_cases = json.loads(result)
            return test_cases.get("test_cases", test_cases)
        except json.JSONDecodeError:
            return result


# ---- 使用示例 ----
if __name__ == "__main__":
    gen = TestCaseGenerator()
    
    requirement = """
    用户注册功能需求：
    1. 用户使用手机号注册，需验证短信验证码
    2. 密码长度8-20位，需包含字母和数字
    3. 同一手机号不可重复注册
    4. 注册成功后自动登录并跳转首页
    5. 验证码60秒内有效，同一号码60秒内不可重复发送
    """
    
    api_spec = """
    POST /api/v1/register
    参数: phone(string, 11位), code(string, 6位), password(string, 8-20位)
    返回: { code: int, message: string, data: { token: string } }
    """
    
    cases = gen.generate(requirement, api_spec)
    
    for case in cases[:5]:
        print(f"[{case['priority']}] {case['title']}")
        print(f"  步骤: {case['steps']}")
        print(f"  预期: {case['expected']}")
        print()

import random
import json

class MockDataGenerator:
    """根据测试意图自动生成Mock测试数据"""
    
    def __init__(self, llm_client):
        self.client = llm_client
    
    def generate(self, test_scenario: str, data_schema: dict, count: int = 10):
        """根据测试场景描述生成针对性的测试数据"""
        
        schema_str = json.dumps(data_schema, ensure_ascii=False, indent=2)
        
        prompt = f"""请为以下测试场景生成{count}条测试数据。

测试场景：{test_scenario}

数据结构：
{schema_str}

要求：
1. 数据必须符合给定的数据结构和类型约束
2. 针对测试场景覆盖不同情况（正常值、边界值、异常值）
3. 金额、日期等字段要合理且有多样性
4. 以JSON数组格式输出"""

        response = self.client.chat.completions.create(
            model="Qwen/Qwen2.5-7B-Instruct",
            messages=[
                {"role": "user", "content": prompt}
            ],
            temperature=0.5,
            response_format={"type": "json_object"},
        )
        
        return json.loads(response.choices[0].message.content).get("data", [])


# 使用示例
schema = {
    "order_id": "string",
    "user_id": "string", 
    "amount": "float(0.01-99999.99)",
    "status": "enum: pending/paid/shipped/delivered/cancelled",
    "created_at": "datetime",
    "items": "array of {sku_id, quantity, price}"
}

mock_gen = MockDataGenerator(OpenAI(base_url="http://localhost:8000/v1", api_key="empty"))

# 场景1：正常订单
normal_orders = mock_gen.generate("正常下单流程的各种状态订单", schema, count=10)

# 场景2：异常金额订单（测试边界）
edge_orders = mock_gen.generate("金额边界值测试：极小值、极大值、0元、负数", schema, count=5)

from openai import OpenAI
import os

class SecurityCodeScanner:
    """AI驱动的代码安全审计工具"""
    
    # 常见漏洞类型
    VULN_TYPES = {
        "sql_injection": "SQL注入",
        "xss": "跨站脚本攻击(XSS)",
        "command_injection": "命令注入",
        "path_traversal": "路径穿越",
        "hardcoded_secret": "硬编码密钥/密码",
        "insecure_deserialization": "不安全的反序列化",
        "open_redirect": "开放重定向",
        "ssrf": "服务端请求伪造(SSRF)",
    }
    
    def __init__(self):
        self.client = OpenAI(base_url="http://localhost:8000/v1", api_key="empty")
    
    def scan_file(self, file_path: str):
        """扫描单个代码文件"""
        with open(file_path, 'r', encoding='utf-8') as f:
            code = f.read()
        
        system_prompt = """你是一位资深的网络安全审计专家，精通OWASP Top 10漏洞类型。

请审计以下代码，识别潜在的安全漏洞。对每个发现的漏洞，提供：
1. 漏洞类型（如SQL注入、XSS等）
2. 风险等级（Critical/High/Medium/Low）
3. 代码位置（行号或函数名）
4. 漏洞描述
5. 修复建议（给出修复后的代码片段）

如果没有发现漏洞，明确说明"未发现明显安全问题"。
以JSON格式输出审计报告。"""

        user_prompt = f"""文件路径: {file_path}
语言: {self._detect_language(file_path)}

代码内容:

response = self.client.chat.completions.create(
            model="Qwen/Qwen2.5-7B-Instruct",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            temperature=0.1,
            max_tokens=4096,
            response_format={"type": "json_object"},
        )
        
        return response.choices[0].message.content
    
    def scan_directory(self, dir_path: str, extensions=None):
        """批量扫描目录下的代码文件"""
        if extensions is None:
            extensions = ['.py', '.java', '.js', '.ts', '.go', '.php', '.rb']
        
        results = []
        for root, _, files in os.walk(dir_path):
            for file in files:
                if any(file.endswith(ext) for ext in extensions):
                    file_path = os.path.join(root, file)
                    report = self.scan_file(file_path)
                    results.append({
                        "file": file_path,
                        "report": report
                    })
                    print(f"✅ 扫描完成: {file_path}")
        return results
    
    def _detect_language(self, file_path: str):
        ext_map = {'.py': 'Python', '.java': 'Java', '.js': 'JavaScript', 
                   '.ts': 'TypeScript', '.go': 'Go', '.php': 'PHP'}
        _, ext = os.path.splitext(file_path)
        return ext_map.get(ext, 'Unknown')


# 使用示例
scanner = SecurityCodeScanner()
report = scanner.scan_file("./src/api/user_login.py")
print(report)

import numpy as np
from sklearn.ensemble import IsolationForest
from collections import deque
import time

class SecurityLogMonitor:
    """基于孤立森林的安全日志异常检测"""
    
    def __init__(self, window_size=10000):
        self.model = IsolationForest(
            n_estimators=100,
            contamination=0.02,    # 预估异常比例2%
            random_state=42,
        )
        self.feature_buffer = deque(maxlen=window_size)
        self.trained = False
        
        # 异常类型分类器（可接LLM做进一步分析）
        self.anomaly_types = {
            "brute_force": "暴力破解",
            "scan_attack": "端口/目录扫描",
            "sql_injection": "SQL注入尝试",
            "xss_attack": "XSS攻击尝试",
            "abnormal_frequency": "请求频率异常",
        }
    
    def extract_features(self, log_entry: dict):
        """从日志条目中提取特征"""
        features = []
        
        # 请求频率特征
        features.append(log_entry.get("request_count_1min", 0))
        features.append(log_entry.get("request_count_5min", 0))
        features.append(log_entry.get("request_count_1hour", 0))
        
        # 请求特征
        url = log_entry.get("url", "")
        features.append(len(url))                           # URL长度
        features.append(url.count("/"))                     # 路径深度
        features.append(int("'" in url or '"' in url))      # 是否含引号(注入特征)
        features.append(int("script" in url.lower()))       # 是否含script(XSS特征)
        features.append(int("union" in url.lower()))        # UNION注入特征
        features.append(int("../" in url))                  # 路径穿越特征
        
        # 响应特征
        features.append(log_entry.get("status_code", 200))
        features.append(log_entry.get("response_time_ms", 0))
        features.append(log_entry.get("response_size", 0))
        
        return np.array(features)
    
    def train(self, historical_logs: list):
        """用历史正常日志训练基线模型"""
        print(f"训练样本数: {len(historical_logs)}")
        features = np.array([self.extract_features(log) for log in historical_logs])
        self.model.fit(features)
        self.trained = True
        print("基线模型训练完成")
    
    def detect(self, log_entry: dict):
        """实时检测单条日志是否异常"""
        if not self.trained:
            raise RuntimeError("模型未训练，请先调用train()")
        
        features = self.extract_features(log_entry).reshape(1, -1)
        is_anomaly = self.model.predict(features)[0] == -1
        anomaly_score = self.model.score_samples(features)[0]
        
        result = {
            "timestamp": log_entry.get("timestamp"),
            "ip": log_entry.get("client_ip"),
            "url": log_entry.get("url"),
            "is_anomaly": is_anomaly,
            "anomaly_score": float(anomaly_score),
            "severity": self._get_severity(anomaly_score),
        }
        
        if is_anomaly:
            result["possible_type"] = self._classify_anomaly(log_entry)
            result["action"] = "ALERT" if result["severity"] == "HIGH" else "LOG"
        
        return result
    
    def _get_severity(self, score):
        if score < -0.7:
            return "HIGH"
        elif score < -0.5:
            return "MEDIUM"
        return "LOW"
    
    def _classify_anomaly(self, log_entry):
        """简单分类异常类型（生产环境可接LLM做更精细分类）"""
        url = log_entry.get("url", "").lower()
        if "union" in url or "'" in url:
            return self.anomaly_types["sql_injection"]
        elif "script" in url:
            return self.anomaly_types["xss_attack"]
        elif "../" in url:
            return self.anomaly_types.get("path_traversal", "路径穿越")
        elif log_entry.get("request_count_1min", 0) > 100:
            return self.anomaly_types["brute_force"]
        else:
            return "未知异常类型"


# 使用示例
monitor = SecurityLogMonitor()

# 模拟历史日志（正常流量）
normal_logs = [
    {"url": "/api/users/123", "status_code": 200, "client_ip": "192.168.1.10",
     "request_count_1min": 5, "request_count_5min": 20, "request_count_1hour": 100,
     "response_time_ms": 50, "response_size": 1024}
    for _ in range(1000)
]
monitor.train(normal_logs)

# 实时检测一条可疑请求
suspicious_log = {
    "timestamp": "2026-08-13T10:30:00",
    "url": "/api/users/1' UNION SELECT * FROM passwords--",
    "status_code": 500,
    "client_ip": "10.0.0.99",
    "request_count_1min": 1,
    "request_count_5min": 3,
    "request_count_1hour": 5,
    "response_time_ms": 200,
    "response_size": 0,
}

result = monitor.detect(suspicious_log)
print(f"异常检测: {result}")
# 输出: is_anomaly=True, possible_type=SQL注入尝试, severity=HIGH

from openai import OpenAI
import sqlite3
import json

class NaturalLanguageQueryEngine:
    """自然语言查询引擎：自然语言 → SQL → 结果"""
    
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self.client = OpenAI(base_url="http://localhost:8000/v1", api_key="empty")
        self.schema = self._get_db_schema()
    
    def _get_db_schema(self):
        """自动提取数据库Schema作为LLM上下文"""
        cursor = self.conn.cursor()
        
        # 获取所有表结构
        cursor.execute("SELECT name, sql FROM sqlite_master WHERE type='table'")
        tables = cursor.fetchall()
        
        schema_info = []
        for table_name, create_sql in tables:
            # 获取列信息
            cursor.execute(f"PRAGMA table_info({table_name})")
            columns = cursor.fetchall()
            
            col_info = []
            for col in columns:
                col_info.append({
                    "name": col[1],
                    "type": col[2],
                    "nullable": not col[3],
                    "primary_key": bool(col[5]),
                })
            
            # 获取样例数据（帮助LLM理解数据格式）
            cursor.execute(f"SELECT * FROM {table_name} LIMIT 3")
            sample_rows = cursor.fetchall()
            sample_data = [dict(zip([c[1] for c in columns], row)) for row in sample_rows]
            
            schema_info.append({
                "table": table_name,
                "columns": col_info,
                "create_sql": create_sql,
                "sample_data": sample_data,
            })
        
        return schema_info
    
    def nl_to_sql(self, question: str):
        """将自然语言问题转换为SQL"""
        
        schema_str = json.dumps(self.schema, ensure_ascii=False, indent=2)
        
        system_prompt = """你是一位SQL专家。请根据用户的自然语言问题，生成正确的SQLite SQL查询。

规则：
1. 只返回一条SELECT语句，不要包含任何解释文字
2. 只允许查询操作，禁止INSERT/UPDATE/DELETE/DROP
3. 使用正确的表名和列名（参考给定的Schema）
4. 如需聚合，使用合理的别名
5. 如需条件过滤，使用参数化方式防止注入（用?占位）

数据库Schema:
""" + schema_str

        response = self.client.chat.completions.create(
            model="Qwen/Qwen2.5-7B-Instruct",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": question},
            ],
            temperature=0.0,    # SQL生成要确定性，温度设0
            max_tokens=1024,
        )
        
        sql = response.choices[0].message.content.strip()
        # 清理可能的markdown标记
        sql = sql.replace("```sql", "").replace("```", "").strip()
        return sql
    
    def validate_sql(self, sql: str):
        """安全校验：只允许SELECT"""
        sql_upper = sql.upper().strip()
        forbidden = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "TRUNCATE", "ATTACH", "DETACH"]
        for kw in forbidden:
            if kw in sql_upper:
                return False, f"检测到危险操作: {kw}"
        if not sql_upper.startswith("SELECT"):
            return False, "只允许SELECT查询"
        return True, "OK"
    
    def query(self, question: str):
        """完整流程：自然语言 → SQL → 执行 → 结果"""
        # 1. 生成SQL
        sql = self.nl_to_sql(question)
        print(f"生成的SQL: {sql}")
        
        # 2. 安全校验
        is_safe, msg = self.validate_sql(sql)
        if not is_safe:
            return {"error": f"安全校验失败: {msg}", "sql": sql}
        
        # 3. 执行查询
        try:
            cursor = self.conn.cursor()
            cursor.execute(sql)
            columns = [desc[0] for desc in cursor.description]
            rows = cursor.fetchall()
            
            results = [dict(zip(columns, row)) for row in rows]
            return {
                "question": question,
                "sql": sql,
                "row_count": len(results),
                "data": results[:100],  # 限制返回100条
            }
        except Exception as e:
            return {"error": str(e), "sql": sql}


# ---- 完整使用示例 ----
if __name__ == "__main__":
    # 假设有一个电商数据库
    engine = NaturalLanguageQueryEngine("./data/ecommerce.db")
    
    # 业务同学直接问
    questions = [
        "上个月销售额最高的前10个商品是什么？",
        "最近一周每天的新增用户数是多少？",
        "退货率超过5%的商品类别有哪些？",
        "每个地区的平均客单价是多少，按从高到低排列",
    ]
    
    for q in questions:
        print(f"\n{'='*60}")
        print(f"问题: {q}")
        result = engine.query(q)
        print(f"SQL: {result.get('sql', '')}")
        print(f"结果: {result.get('row_count', 0)} 条记录")
        if result.get('data'):
            print(f"前3条: {json.dumps(result['data'][:3], ensure_ascii=False)}")

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from datetime import datetime, timedelta

class DataQualityMonitor:
    """AI驱动的数据质量监控"""
    
    def __init__(self):
        self.baseline_models = {}   # 每张表一个基线模型
    
    def build_baseline(self, table_name: str, df: pd.DataFrame):
        """为每张表建立正常数据基线"""
        # 提取统计特征
        features = self._extract_features(df)
        
        model = IsolationForest(contamination=0.05, random_state=42)
        model.fit(features)
        
        self.baseline_models[table_name] = {
            "model": model,
            "columns": df.columns.tolist(),
            "dtypes": df.dtypes.to_dict(),
            "stats": {
                col: {
                    "mean": df[col].mean() if df[col].dtype in ['int64', 'float64'] else None,
                    "std": df[col].std() if df[col].dtype in ['int64', 'float64'] else None,
                    "null_ratio": df[col].isnull().mean(),
                    "unique_ratio": df[col].nunique() / len(df) if len(df) > 0 else 0,
                }
                for col in df.columns
            }
        }
        print(f"✅ 表 {table_name} 基线模型已建立")
    
    def _extract_features(self, df: pd.DataFrame):
        """提取数据质量特征"""
        features = []
        
        for col in df.columns:
            # 数值型特征
            if df[col].dtype in ['int64', 'float64']:
                features.append(df[col].mean())
                features.append(df[col].std())
                features.append(df[col].min())
                features.append(df[col].max())
            
            # 通用特征
            features.append(df[col].isnull().mean())        # 空值率
            features.append(df[col].nunique() / max(len(df), 1))  # 唯一值率
        
        # 行数特征
        features.append(len(df))
        
        return np.array(features).reshape(1, -1)
    
    def check(self, table_name: str, df: pd.DataFrame):
        """检测当前数据批次是否偏离基线"""
        if table_name not in self.baseline_models:
            return {"status": "no_baseline", "message": "该表未建立基线"}
        
        baseline = self.baseline_models[table_name]
        baseline_stats = baseline["stats"]
        
        issues = []
        
        # 1. 检查列结构变化
        if set(df.columns) != set(baseline["columns"]):
            added = set(df.columns) - set(baseline["columns"])
            removed = set(baseline["columns"]) - set(df.columns)
            issues.append({
                "type": "schema_change",
                "severity": "HIGH",
                "message": f"表结构变化: 新增列{added}, 删除列{removed}",
            })
        
        # 2. 逐列检查统计特征偏移
        for col in df.columns:
            if col not in baseline_stats:
                continue
            
            b_stats = baseline_stats[col]
            
            # 空值率异常
            current_null_ratio = df[col].isnull().mean()
            if current_null_ratio > b_stats["null_ratio"] * 3 and current_null_ratio > 0.1:
                issues.append({
                    "type": "null_ratio_anomaly",
                    "severity": "HIGH",
                    "column": col,
                    "message": f"空值率异常: 基线{b_stats['null_ratio']:.2%} → 当前{current_null_ratio:.2%}",
                })
            
            # 数值分布偏移
            if b_stats["mean"] is not None and b_stats["std"] is not None:
                current_mean = df[col].mean()
                if abs(current_mean - b_stats["mean"]) > 3 * b_stats["std"]:
                    issues.append({
                        "type": "distribution_shift",
                        "severity": "MEDIUM",
                        "column": col,
                        "message": f"均值偏移: 基线{b_stats['mean']:.2f} → 当前{current_mean:.2f}",
                    })
        
        # 3. 行数异常
        baseline_rows = baseline["stats"].get("_row_count", 0)
        current_rows = len(df)
        if baseline_rows > 0:
            change_ratio = abs(current_rows - baseline_rows) / baseline_rows
            if change_ratio > 0.3:
                issues.append({
                    "type": "volume_anomaly",
                    "severity": "HIGH" if change_ratio > 0.5 else "MEDIUM",
                    "message": f"数据量异常: 基线{baseline_rows}行 → 当前{current_rows}行 (变化{change_ratio:.0%})",
                })
        
        return {
            "table": table_name,
            "timestamp": datetime.now().isoformat(),
            "status": "PASS" if not issues else "FAIL",
            "issues": issues,
            "current_rows": current_rows,
        }


# 使用示例
monitor = DataQualityMonitor()

# 用历史数据建立基线
baseline_data = pd.read_sql("SELECT * FROM orders WHERE date < '2026-08-01'", conn)
monitor.build_baseline("orders", baseline_data)

# 检查今天的新数据
today_data = pd.read_sql("SELECT * FROM orders WHERE date = '2026-08-13'", conn)
report = monitor.check("orders", today_data)

if report["status"] == "FAIL":
    print("⚠️ 数据质量告警！")
    for issue in report["issues"]:
        print(f"  [{issue['severity']}] {issue['type']}: {issue['message']}")