# 创建项目目录
mkdir php-ai-data-generator && cd php-ai-data-generator# 安装Python环境
python -m venv venv
source venv/bin/activate
pip install faker transformers redis minio# 安装PHP依赖
composer create-project laravel/laravel api
cd api
composer require predis/predis league/flysystem-aws-s3-v3

import json
import redis
from faker import Faker
from transformers import pipeline
from minio import Minioclass DataGenerator:def __init__(self):self.fake = Faker('zh_CN')self.text_generator = pipeline('text-generation', model='uer/gpt2-chinese-cluecorpussmall')# 连接Redisself.redis = redis.Redis(host='localhost', port=6379, db=0)# 连接MinIOself.minio = Minio("localhost:9000",access_key="your-access-key",secret_key="your-secret-key",secure=False)def generate_user_data(self, count):"""生成用户结构化数据"""users = []for _ in range(count):user = {'id': self.fake.uuid4(),'name': self.fake.name(),'email': self.fake.email(),'phone': self.fake.phone_number(),'address': self.fake.address(),'company': self.fake.company(),'job': self.fake.job(),'birthdate': str(self.fake.date_of_birth())}users.append(user)return usersdef generate_product_reviews(self, count, product_name):"""生成商品评论文本"""reviews = []for _ in range(count):prompt = f"作为消费者，我对{product_name}的评价是："review = self.text_generator(prompt, max_length=100,num_return_sequences=1,temperature=0.7)[0]['generated_text']reviews.append({'product': product_name,'review': review.replace(prompt, '').strip(),'rating': self.fake.random_int(1, 5)})return reviewsdef process_queue(self):"""监听Redis队列并处理任务"""while True:# 从队列获取任务task_data = self.redis.brpop('data_tasks', timeout=30)if not task_data:continuetask = json.loads(task_data[1])print(f"Processing task: {task['id']}")try:# 执行生成任务if task['type'] == 'user_data':data = self.generate_user_data(task['count'])elif task['type'] == 'product_review':data = self.generate_product_reviews(task['count'], task['product_name'])# 保存到MinIOfilename = f"{task['type']}_{task['id']}.json"data_bytes = json.dumps(data, ensure_ascii=False).encode('utf-8')self.minio.put_object("datasets",filename,io.BytesIO(data_bytes),length=len(data_bytes),content_type='application/json')# 更新任务状态self.redis.set(f"task:{task['id']}:status", "completed")self.redis.set(f"task:{task['id']}:file", filename)except Exception as e:print(f"Error processing task: {str(e)}")self.redis.set(f"task:{task['id']}:status", "failed")

from core import DataGeneratorif __name__ == "__main__":print("Starting AI data generator worker...")generator = DataGenerator()generator.process_queue()

class CreateTasksTable extends Migration
{public function up(){Schema::create('tasks', function (Blueprint $table) {$table->id();$table->string('type'); // user_data, product_review$table->integer('count');$table->string('status')->default('pending'); // pending, processing, completed, failed$table->string('file_path')->nullable();$table->json('params')->nullable(); // 额外参数$table->timestamps();});}
}

class Task extends Model
{protected $casts = ['params' => 'array'];
}

class TaskController extends Controller
{public function create(Request $request){$validated = $request->validate(['type' => 'required|in:user_data,product_review','count' => 'required|integer|min:10|max:100000','product_name' => 'required_if:type,product_review']);$task = Task::create(['type' => $validated['type'],'count' => $validated['count'],'params' => $request->except(['type', 'count'])]);// 发送到Redis队列Redis::rpush('data_tasks', json_encode(['id' => $task->id,'type' => $task->type,'count' => $task->count,'product_name' => $task->params['product_name'] ?? null]));return response()->json(['message' => '任务创建成功','task_id' => $task->id]);}public function status($id){$task = Task::findOrFail($id);$redisStatus = Redis::get("task:{$id}:status");return response()->json(['status' => $redisStatus ?? $task->status,'file_url' => $task->file_path ? Storage::disk('minio')->temporaryUrl($task->file_path,now()->addMinutes(30): null]);}
}

class ProcessTaskCompletion implements ShouldQueue
{use Dispatchable, InteractsWithQueue;public function __construct(public int $taskId) {}public function handle(){$task = Task::find($this->taskId);// 从Redis获取结果$filename = Redis::get("task:{$this->taskId}:file");$status = Redis::get("task:{$this->taskId}:status");if ($status === 'completed' && $filename) {$task->update(['status' => 'completed','file_path' => $filename]);// 发送通知event(new TaskCompleted($task));} else {$task->update(['status' => 'failed']);}}
}

class Kernel extends ConsoleKernel
{protected function schedule(Schedule $schedule){// 每分钟检查任务状态$schedule->call(function () {Task::where('status', 'pending')->where('created_at', '>', now()->subHours(24))->each(function ($task) {ProcessTaskCompletion::dispatch($task->id);});})->everyMinute();}
}

<template><div class="task-creator"><h3>创建数据生成任务</h3><div class="form-group"><label>数据类型</label><select v-model="form.type" @change="onTypeChange"><option value="user_data">用户数据</option><option value="product_review">商品评论</option></select></div><div class="form-group"><label>生成数量</label><input type="range" min="100" max="10000" step="100"v-model="form.count"><span>{{ form.count }} 条</span></div><div v-if="form.type === 'product_review'" class="form-group"><label>商品名称</label><input type="text" placeholder="输入商品名称"v-model="form.product_name"></div><button @click="submit">开始生成</button><div v-if="taskId" class="task-status"><p>任务ID: {{ taskId }}</p><div class="progress-bar"><div :style="{ width: progress + '%' }"></div></div><p v-if="downloadUrl"><a :href="downloadUrl" download>下载生成数据</a></p></div></div>
</template><script>
import { ref, watch } from 'vue';
import axios from 'axios';export default {setup() {const form = ref({type: 'user_data',count: 1000,product_name: ''});const taskId = ref(null);const progress = ref(0);const downloadUrl = ref(null);const submit = async () => {try {const response = await axios.post('/api/tasks', form.value);taskId.value = response.data.task_id;// 轮询任务状态const interval = setInterval(async () => {const statusRes = await axios.get(`/api/tasks/${taskId.value}/status`);if (statusRes.data.status === 'processing') {progress.value = 50;} else if (statusRes.data.status === 'completed') {progress.value = 100;downloadUrl.value = statusRes.data.file_url;clearInterval(interval);} else if (statusRes.data.status === 'failed') {alert('任务处理失败');clearInterval(interval);}}, 3000);} catch (error) {alert(`创建失败: ${error.response.data.message}`);}};return { form, submit, taskId, progress, downloadUrl };}
};
</script><style scoped>
.progress-bar {height: 20px;background: #eee;border-radius: 10px;overflow: hidden;
}
.progress-bar > div {height: 100%;background: #42b983;transition: width 0.3s ease;
}
</style>

<template><div class="dashboard"><div class="stats"><div class="stat-card"><h4>今日生成数据</h4><p>{{ stats.today_count }} 条</p></div><div class="stat-card"><h4>任务成功率</h4><p>{{ stats.success_rate }}%</p></div><div class="stat-card"><h4>平均生成时间</h4><p>{{ stats.avg_time }} 秒</p></div></div><h3>最近任务</h3><table><thead><tr><th>ID</th><th>类型</th><th>数量</th><th>状态</th><th>创建时间</th></tr></thead><tbody><tr v-for="task in recentTasks" :key="task.id"><td>{{ task.id }}</td><td>{{ task.type }}</td><td>{{ task.count }}</td><td :class="task.status">{{ task.status }}</td><td>{{ formatDate(task.created_at) }}</td></tr></tbody></table></div>
</template><script>
import { ref, onMounted } from 'vue';
import axios from 'axios';export default {setup() {const stats = ref({today_count: 0,success_rate: 0,avg_time: 0});const recentTasks = ref([]);const fetchData = async () => {const [statsRes, tasksRes] = await Promise.all([axios.get('/api/stats'),axios.get('/api/tasks?limit=5')]);stats.value = statsRes.data;recentTasks.value = tasksRes.data;};onMounted(() => {fetchData();setInterval(fetchData, 30000); // 每30秒刷新});return { stats, recentTasks };}
};
</script>

# 启动Redis
docker run -p 6379:6379 redis# 启动MinIO
docker run -p 9000:9000 minio/minio server /data# 启动Python工作线程
python generator/run_worker.py# 启动Laravel
cd api && php artisan serve --port=8000# 启动Vue前端
cd frontend && npm run dev

// 测试API请求
const response = await axios.post('http://localhost:8000/api/tasks', {type: 'product_review',count: 500,product_name: '智能手机'
});console.log('任务ID:', response.data.task_id);// 轮询任务状态
const checkStatus = async (taskId) => {const res = await axios.get(`http://localhost:8000/api/tasks/${taskId}/status`);console.log('任务状态:', res.data.status);if (res.data.status === 'completed') {console.log('下载链接:', res.data.file_url);} else if (res.data.status !== 'failed') {setTimeout(() => checkStatus(taskId), 3000);}
};checkStatus(response.data.task_id);

# 使用批量生成代替单条生成
def generate_batch_users(count, batch_size=1000):users = []for _ in range(0, count, batch_size):batch = [self._generate_user() for _ in range(min(batch_size, count))]users.extend(batch)count -= batch_sizereturn users

# 添加数据验证规则
def validate_user(user):required_fields = ['name', 'email', 'phone']return all(field in user and user[field] for field in required_fields)# 在生成后添加验证
valid_users = [user for user in users if validate_user(user)]

// Laravel异常处理器
public function report(Exception $exception)
{if ($exception instanceof DataGenerationException) {Log::channel('data_generation')->error($exception);Redis::set("task:{$taskId}:status", "failed");}parent::report($exception);
}

# 安装eksctl（AWS官方管理工具）
brew tap weaveworks/tap
brew install weaveworks/tap/eksctl# 创建集群配置文件 cluster.yaml
apiVersion: eksctl.io/v1alpha5
kind: ClusterConfig
metadata:name: ai-data-clusterregion: ap-northeast-1
nodeGroups:- name: ng-1instanceType: m6i.largedesiredCapacity: 3volumeSize: 100# 启动集群（约15分钟）
eksctl create cluster -f cluster.yaml

# PHP-FPM容器
FROM php:8.2-fpm-alpine
RUN docker-php-ext-install pdo_mysql sockets
COPY --from=composer /usr/bin/composer /usr/bin/composer
WORKDIR /var/www# Python AI容器
FROM python:3.11-slim
RUN pip install torch transformers faker redis minio
COPY generator/ /app
CMD ["python", "/app/run_worker.py"]

# hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:name: py-generator-hpa
spec:scaleTargetRef:apiVersion: apps/v1kind: Deploymentname: py-generatorminReplicas: 3maxReplicas: 20metrics:- type: Resourceresource:name: cputarget:type: UtilizationaverageUtilization: 70

events {worker_connections 10000;  # 提升单节点并发能力
}
http {proxy_buffering off;       # 禁用缓冲提升实时性keepalive_timeout 65;      # 减少TCP握手开销
}

class VerifyJWT
{public function handle($request, Closure $next) {try {$user = JWTAuth::parseToken()->authenticate();if (!$user) throw new Exception('User not found');} catch (Exception $e) {return response()->json(['error' => 'Unauthorized'], 401);}return $next($request);}
}

# Windows终端操作
# 查看所有Python版本
py -0# 为3.6创建专用别名
setx PYTHON36 "C:\Python36\python.exe"# 运行特定版本
%PYTHON36% generator.py

# Linux使用update-alternatives
sudo update-alternatives --install /usr/bin/python3 python3 /usr/bin/python3.6 1
sudo update-alternatives --install /usr/bin/python3 python3 /usr/bin/python3.11 2# 切换版本
sudo update-alternatives --config python3

# 替换通用模型为中文优化模型
generator = pipeline('text-generation', model='uer/gpt2-chinese-cluecorpussmall',tokenizer='bert-base-chinese')# 添加领域关键词增强
keywords = ["电商", "用户", "购买", "好评"]
prompt += "。请使用以下关键词生成文本：" + "、".join(keywords)

from minio.error import S3Errordef stream_to_minio(bucket, filename, data_generator):try:with io.BytesIO() as buffer:for chunk in data_generator:  # 每次生成100条buffer.write(json.dumps(chunk).encode('utf-8'))if buffer.tell() > 10*1024*1024:  # 每10MB刷新buffer.seek(0)minio_client.put_object(bucket, filename, buffer, length=buffer.tell())buffer = io.BytesIO()# 写入剩余数据if buffer.tell() > 0:buffer.seek(0)minio_client.put_object(bucket, filename, buffer, length=buffer.tell())except S3Error as err:print(f"S3 error: {err}")

# 1. 网络连通性检查
nc -zv 127.0.0.1 6379  # 测试Redis端口# 2. 协议分析（Wireshark过滤条件）
tcp.port == 8000 and http# 3. PHPY集成方案（替代Redis队列）
$os = PyCore::import('os')
$sys = $os->uname()
echo $sys->machine;  # 直接调用Python对象

graph TDA[原始数据] --> B{Presidio分析器}B -->|识别敏感字段| C[匿名引擎]C --> D[替换姓名→**]C --> E[替换电话→138****1234]C --> F[加密地址]D --> G[合规数据]