xinyouhe/useds
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

推送队列

Browse Source
This commit is contained in:
WIN-C27C5MSVTPO\16961
2026-03-11 16:01:18 +08:00
parent 261fa05517
commit ff3201375d
13 changed files with 188 additions and 887 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/common/controller/UsedCarDetailController.php
+10 -3
View File
@@ -268,10 +268,17 @@ class UsedCarDetailController extends BaseController
try {
$data = $request->post();
$this->service->audit($request, $data);
$result = $this->service->audit($request, $data);
$statusText = $data['audit_status'] == 1 ? '通过' : '拒绝';
return $this->success([], "审核{$statusText}成功");
$message = "审核{$statusText}成功";
// 审核通过时，提示推送任务已加入队列
if ($data['audit_status'] == 1 && $result['push_queued']) {
$message .= '，WordPress推送任务已加入队列，将在后台处理';
}
return $this->success($result, $message);
} catch (\Throwable $e) {
return $this->error($e->getMessage());
}
@@ -344,7 +351,7 @@ class UsedCarDetailController extends BaseController
$result = $this->service->crawlAndStore($url, $userId);
return $this->success($result, '抓取成功，已保存为待审核状态');
return $this->success($result, '抓取成功，WordPress推送任务已加入队列，将在后台处理');
} catch (\Throwable $e) {
Log::error('抓取懂车帝二手车失败', [
'url' => $request->post('url'),
app/common/service/UsedCarDetailService.php
+32 -56
View File
@@ -424,70 +424,33 @@ class UsedCarDetailService
// 更新车辆状态
$result = UsedCarDetail::where('id', $carId)->update(['status' => $auditStatus]);
// 审核通过后自动推送到WordPress
$pushResult = ['success' => true, 'message' => ''];
// 审核通过后，将推送任务发送到队列
$pushQueued = false;
if ($auditStatus == 1) {
$pushResult = $this->pushToWordpressAfterAudit($carId);
// 推送失败，将车辆状态修改为3
if (!$pushResult['success']) {
UsedCarDetail::where('id', $carId)->update(['status' => 3]);
try {
// 发送到 Redis 队列异步处理
\Webman\RedisQueue\Redis::send('wordpress-push', [
'car_id' => $carId,
]);
$pushQueued = true;
Log::info('WordPress推送任务已加入队列', ['car_id' => $carId]);
} catch (\Throwable $e) {
Log::error('WordPress推送任务加入队列失败', [
'car_id' => $carId,
'error' => $e->getMessage(),
]);
$pushQueued = false;
}
}
// 返回审核结果和推送结果
// 返回审核结果
return [
'audit_success' => true,
'push_success' => $pushResult['success'],
'push_message' => $pushResult['message'],
'message' => $pushResult['success'] ? '审核通过并推送成功' : '审核通过，但推送失败：' . $pushResult['message'],
'push_queued' => $pushQueued,
'message' => $pushQueued ? '审核通过，WordPress推送任务已加入队列' : '审核通过',
];
}
/**
* 审核通过后推送车辆到WordPress
* @return array ['success' => bool, 'message' => string]
*/
private function pushToWordpressAfterAudit(int $carId): array
{
try {
// 获取启用的WordPress站点
$site = \app\common\model\WordpressSite::where('status', 1)->first();
if (!$site) {
$msg = '没有启用的WordPress站点';
Log::warning('WordPress推送失败：' . $msg, ['car_id' => $carId]);
return ['success' => false, 'message' => $msg];
}
// 检查是否已推送过
$existing = \app\common\model\WordpressUsedCar::where('wp_site_id', $site->id)
->where('used_car_id', $carId)
->where('status', 1)
->first();
// if ($existing) {
// $msg = '车辆已推送，跳过';
// Log::info('WordPress推送：' . $msg, ['car_id' => $carId]);
// return ['success' => true, 'message' => $msg];
// }
// 调用推送服务
$wpService = new \app\common\service\WordpressSiteService();
$wpService->createProductForNewSite($site, $carId);
Log::info('WordPress推送成功', ['car_id' => $carId, 'site_id' => $site->id]);
return ['success' => true, 'message' => ''];
} catch (\Throwable $e) {
$msg = $e->getMessage();
Log::error('WordPress推送失败', [
'car_id' => $carId,
'error' => $msg,
'trace' => $e->getTraceAsString(),
]);
return ['success' => false, 'message' => $msg];
}
}
/**
* 获取审核记录列表
*/
@@ -957,7 +920,7 @@ class UsedCarDetailService
'crawl_year' => $info['year'] ?? '',
'crawl_msrp' => $info['new_price'] ?? '',
'crawl_car_id' => $info['car_id'] ?? 0,
'status' => 0, // 待审核
'status' => 3, // 待推送状态
'images' => $imagesValue,
'power' => $info['horsepower'] ?? '',
];
@@ -971,6 +934,19 @@ class UsedCarDetailService
// 5. 创建记录
$car = UsedCarDetail::create($data);
// 6. 将推送任务发送到队列
try {
\Webman\RedisQueue\Redis::send('wordpress-push', [
'car_id' => $car->id,
]);
Log::info('WordPress推送任务已加入队列', ['car_id' => $car->id]);
} catch (\Throwable $e) {
Log::error('WordPress推送任务加入队列失败', [
'car_id' => $car->id,
'error' => $e->getMessage(),
]);
}
return $car->toArray();
}
app/queue/redis/ApprovalCallback.php
-83
View File
@@ -1,83 +0,0 @@
<?php
namespace app\queue\redis;
use support\exception\BusinessException;
use Webman\RedisQueue\Consumer;
class ApprovalCallback implements Consumer
{
// 要消费的队列名
public $queue = 'approval-callback';
// 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
public function consume($data)
{
try {
match ($data['flow_type']) {
'stock_orders' => $this->approvalStockOrders($data),
'purchase_orders' => $this->approvalPurchaseOrders($data),
'stock_orders_nullify' => '',
default => throw new BusinessException('未知的审批类型，无法执行回调操作')
};
return true;
} catch (\Throwable $th) {
flg('审批回调异常：'.$th->getFile()."|".$th->getLine()."|".$th->getMessage(), false);
throw new BusinessException($th->getFile()."|".$th->getLine()."|".$th->getMessage());
}
}
// 消费失败回调
/*
$package = [
'id' => 1357277951, // 消息ID
'time' => 1709170510, // 消息时间
'delay' => 0, // 延迟时间
'attempts' => 2, // 消费次数
'queue' => 'send-mail', // 队列名
'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
'max_attempts' => 5, // 最大重试次数
'error' => '错误信息' // 错误信息
]
*/
public function onConsumeFailure(\Throwable $e, $package)
{
echo "consume failure\n";
echo $e->getMessage() . "\n";
// 无需反序列化
var_export($package);
}
/**
* 备货单处理
*/
private function approvalStockOrders($data)
{
// 生成PDF文件
$service = new \app\crm\service\StockOrdersService();
$result = $service->createdPdfFile($data['biz_id']);
return $result;
}
/**
* 采购单处理
*/
private function approvalPurchaseOrders($data)
{
$udata['id'] = $data['biz_id'];
if ($data['status'] === 'approve') {
$udata['status'] = 2;
}
if ($data['status'] === 'reject') {
$udata['status'] = 3;
}
$service = new \app\common\service\PurchaseOrdersService();
$result = $service->updateStatus($udata);
return $result;
}
}
app/queue/redis/EmailMessage.php
-51
View File
@@ -1,51 +0,0 @@
<?php
namespace app\queue\redis;
use Webman\RedisQueue\Consumer;
use app\inquiry\service\InquiryLogService;
use app\common\utils\email\MailSend;
class EmailMessage implements Consumer
{
// 要消费的队列名
public $queue = 'email-message';
// 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
public function consume($data)
{
try {
// 询单信息
$inquiryLogService = new InquiryLogService();
$content = $inquiryLogService->getContent($data['id']);
MailSend::send($data['email'], $data['title'], replaceTpl($content));
} catch (\Throwable $th) {
throw $th;
}
}
// 消费失败回调
/*
$package = [
'id' => 1357277951, // 消息ID
'time' => 1709170510, // 消息时间
'delay' => 0, // 延迟时间
'attempts' => 2, // 消费次数
'queue' => 'send-mail', // 队列名
'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
'max_attempts' => 5, // 最大重试次数
'error' => '错误信息' // 错误信息
]
*/
public function onConsumeFailure(\Throwable $e, $package)
{
echo "consume failure\n";
echo $e->getMessage() . "\n";
// 无需反序列化
var_export($package);
}
}
app/queue/redis/FeishuMessage.php
-131
View File
@@ -1,131 +0,0 @@
<?php
namespace app\queue\redis;
use app\common\model\CorpFeishuApp;
use app\common\model\CrmCustomer;
use Webman\RedisQueue\Consumer;
use app\common\model\User;
use app\common\utils\FeishuMessage as FeishuMsg;
use support\exception\BusinessException;
use app\inquiry\service\InquiryLogService;
class FeishuMessage implements Consumer
{
// 要消费的队列名
public $queue = 'feishu-message';
// 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
/**
* @param string $category 分类：crm | inquiry
* @param string $scene 场景：group-send | group-update | person-send | person-update
* @param string $id 数据ID：询单ID | 客户ID
* @param string $receive_type 接收者类型：group | person
*/
public function consume($data)
{
try {
if ($data['category'] === 'crm') {
// 客户线索
$crmCustomer = new CrmCustomer();
$info = $crmCustomer->where('id', $data['id'])->first();
// 跟单人
$reInfo = User::where('id', $info->user_id)->first();
// 飞书应用
$feishuApp = new CorpFeishuApp();
$feishuAppInfo = $feishuApp->where('corp_id', $reInfo->corp_id)->where('system_id', 1)->first();
// 推送数据
$options = [
'contacts' => $info->contacts,
'follow_last' => $info->follow_last ? date('Y-m-d H:i', $info->follow_last) : '',
'follow_next' => $info->follow_next ? date('Y-m-d H:i', $info->follow_next) : '',
'app_id' => $feishuAppInfo->app_id,
'app_secret' => $feishuAppInfo->app_secret,
'state' => $feishuAppInfo->state,
'msg_type' => 'interactive',
'receive_id_type' => 'open_id',
'receive_id' => $reInfo->open_id
];
}
if ($data['category'] === 'inquiry') {
// 询单信息
$inquiryLogService = new InquiryLogService();
$info = $inquiryLogService->getInquiryLogBy($data['id']);
// 接受者处理
if ($data['receive_type'] == 'group') {
$appId = 'cli_a6183b882cb8500d';
$appSecret = 'VqYib48u4EtDIa6nG9qH3f0fYCA1bCvs';
$state = 'hesu';
$receiveId = 'oc_a52d9f89ad76735279088c78beade8dc';
$receiveIdType = 'chat_id';
} else {
// 跟单人
$reInfo = User::where('id', $info->recipient)->first();
// 飞书应用
$feishuApp = new CorpFeishuApp();
$feishuAppInfo = $feishuApp->where('corp_id', $reInfo->corp_id)->where('system_id', 1)->first();
// 推送信息
$appId = $feishuAppInfo->app_id;
$appSecret = $feishuAppInfo->app_secret;
$state = $feishuAppInfo->state;
$receiveId = $reInfo->open_id;
$receiveIdType = 'open_id';
}
// 推送数据
$options = [
'order_no' => $info->order_no,
'quality' => $inquiryLogService->qualityName($info->quality),
'channel' => $info->channelInfo ? $info->channelInfo->name : '',
'country' => $info->countryInfo ? $info->countryInfo->cn_addr : '',
'enter_user' => $info->enter ? $info->enter->name : '',
'r_time' => $info->r_time ? date('Y-m-d H:i:s', $info->r_time) : '',
't_time' => $info->t_time ? date('Y-m-d H:i:s', $info->t_time) : '',
'app_id' => $appId,
'app_secret' => $appSecret,
'state' => $state,
'msg_type' => 'interactive',
'receive_id_type' => $receiveIdType,
'receive_id' => $receiveId
];
}
$feishuMsg = new FeishuMsg($options);
$feishuMsg->messageSend($data['scene'], $options);
} catch (\Throwable $th) {
throw $th;
}
}
// 消费失败回调
/*
$package = [
'id' => 1357277951, // 消息ID
'time' => 1709170510, // 消息时间
'delay' => 0, // 延迟时间
'attempts' => 2, // 消费次数
'queue' => 'send-mail', // 队列名
'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
'max_attempts' => 5, // 最大重试次数
'error' => '错误信息' // 错误信息
]
*/
public function onConsumeFailure(\Throwable $e, $package)
{
echo "consume failure\n";
echo $e->getMessage() . "\n";
// 无需反序列化
var_export($package);
}
}
app/queue/redis/InquiryToCrmNew.php
-222
View File
@@ -1,222 +0,0 @@
<?php
namespace app\queue\redis;
use app\inquiry\service\InquiryLogService;
use app\common\model\CrmCustomer;
use app\common\model\InquiryLog;
use Webman\RedisQueue\Consumer;
use app\common\model\User;
use app\common\utils\FeishuMessage;
use support\exception\BusinessException;
use support\Log;
use support\Model;
class InquiryToCrmNew implements Consumer
{
// 要消费的队列名
public $queue = 'inquiry-to-crm-new';
// 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
public function consume($data)
{
/*
询单 与 CRM 之间的同步场景描述
添加（询单）
添加 不发 不分 （不同步）1
添加 发了 不分 （要同步）2
添加 不发 分了 （要同步）3
添加 不发 直分 （要同步）4
更新（询单）
没发 没分 更新 （不同步）5
没发 分过 更新 （要同步）6
发过 没分 更新 （要同步）7
发过 分过 更新 （要同步）8
*/
try {
// 获取询盘信息
$info = InquiryLog::with(['channelInfo', 'countryInfo', 'enter'])->where('id', $data['id'])->first();
// 无论添加 或 更新，询单必然要存在
if (!$info) {
throw new BusinessException('询盘不存在：' . $data['id']);
}
if ($info->status == 0 && $info->recipient == 0) {
// 如果询单未发、未分，则不与crm同步，不做任何处理
return true;
}
// 更改库信息
$this->syncCustomer($info);
} catch (\Throwable $th) {
throw $th;
}
}
public function pushWs($title, $content, $recipient = [])
{
$wsdata = [
'sender_id' => 0,
'sender_type' => 'system',
'title' => $title,
'content' => $content,
'scene' => 1,
'recipient' => $recipient
];
// 异步推送
\Webman\RedisQueue\Redis::send('notice', $wsdata);
}
public function pushEmail($id, $title, $email)
{
$options = [
'id' => $id,
'email' => $email,
'title' => $title,
];
\Webman\RedisQueue\Redis::send('email-message', $options);
}
public function syncCustomer(Model $info)
{
$uInfo = $this->recipientInfo($info);
$crmCustomerInfo = CrmCustomer::where('sn', $info->order_no)->first();
if ($crmCustomerInfo) {
// 消息推送
if ($info->status == 2 && empty($crmCustomerInfo->user_id)) {
// $notice = '分配';
$title = '[询单·分配] 新的客户线索【' . $info->order_no . '】，请及时处理';
$content = '[询单·分配] 新的客户线索【' . $info->order_no . '】，请及时处理';
$recipient = [$info->recipient];
// $this->pushWs($title, $content, $recipient);
$this->pushEmail($info->id, $title, $uInfo['uinfo']['email']);
} else {
if ($info->status == 2 && $crmCustomerInfo->user_id == $info->recipient) {
// $notice = '更新';
$title = '[询单·更新] 客户线索' . $info->order_no . '有新的调整，请及时处理';
$content = '[询单·更新] 客户线索' . $info->order_no . '有新的调整，请及时处理';
$recipient = [$info->recipient];
// $this->pushWs($title, $content, $recipient);
$this->pushEmail($info->id, $title, $uInfo['uinfo']['email']);
} else {
// $notice = '重新分配';
$title = '[询单·分配] 新的客户线索【' . $info->order_no . '】，请及时处理';
$content = '[询单·分配] 新的客户线索【' . $info->order_no . '】，请及时处理';
$recipient = [$info->recipient];
// $this->pushWs($title, $content, $recipient);
$this->pushEmail($info->id, $title, $uInfo['uinfo']['email']);
}
}
$udata = [
'corp_id' => $uInfo['corpId'],
'user_id' => $info->recipient,
'channel_id' => $info->channel,
'country_id' => $info->country,
'magnitude' => $info->quality,
'email' => $info->email,
'phone' => $info->tel,
'WhatsApp' => $info->whatsapp,
'contacts' => $info->contacts,
'desc_collect' => $info->content,
];
if (!$crmCustomerInfo->update($udata)) {
throw new BusinessException('同步失败，更新客户线索失败');
}
} else {
$crmCustomer = new CrmCustomer();
$crmCustomer->sn = $info->order_no;
$crmCustomer->corp_id = $uInfo['corpId'];
$crmCustomer->user_id = $uInfo['recipient'];
$crmCustomer->type = 1;
$crmCustomer->channel_id = $info->channel;
$crmCustomer->country_id = $info->country;
$crmCustomer->magnitude = $info->quality;
$crmCustomer->email = $info->email;
$crmCustomer->phone = $info->tel;
$crmCustomer->WhatsApp = $info->whatsapp;
$crmCustomer->contacts = $info->contacts;
$crmCustomer->desc_collect = $info->content;
$crmCustomer->enter_user_id = $info->enter_user_id;
$crmCustomer->obtain = time();
if(empty($uInfo['recipient'])){
$crmCustomer->deleted_at = time();
}
// 修改同步CRM状态
$info->sync_crm = 1;
// 保存
if (!$crmCustomer->save() || !$info->save()) {
throw new BusinessException('同步失败，创建客户线索失败');
}
// 消息推送
if ($info->status == 1) {
// $notice = '发布';
$title = '[询单·公海] 新的客户线索【' . $info->order_no . '】发布';
$content = '[询单·公海] 新的客户线索【' . $info->order_no . '】发布';
$recipient = [$info->recipient];
// $this->pushWs($title, $content);
$this->pushEmail($info->id, $title, $uInfo['uinfo']['email']);
}
if ($info->status == 2) {
// $notice = '分配';
$title = '[询单·分配] 新的客户线索【' . $info->order_no . '】，请及时处理';
$content = '[询单·分配] 新的客户线索【' . $info->order_no . '】，请及时处理';
$recipient = [$info->recipient];
// $this->pushWs($title, $content, $recipient);
$this->pushEmail($info->id, $title, $uInfo['uinfo']['email']);
}
}
return true;
}
public function recipientInfo(Model $info)
{
$recipient = 0;
$corpId = 0;
$uinfo = null;
if (!empty($info->recipient)) {
$recipient = $info->recipient;
$uinfo = User::where('id', $info->recipient)->first();
if ($uinfo) {
$corpId = $uinfo->corp_id;
}
}
return [
'recipient' => $recipient,
'corpId' => $corpId,
'uinfo' => $uinfo
];
}
// 消费失败回调
/*
$package = [
'id' => 1357277951, // 消息ID
'time' => 1709170510, // 消息时间
'delay' => 0, // 延迟时间
'attempts' => 2, // 消费次数
'queue' => 'send-mail', // 队列名
'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
'max_attempts' => 5, // 最大重试次数
'error' => '错误信息' // 错误信息
]
*/
public function onConsumeFailure(\Throwable $e, $package)
{
echo "consume failure\n";
echo $e->getMessage() . "\n";
// 无需反序列化
var_export($package);
}
}
app/queue/redis/Notice.php
-61
View File
@@ -1,61 +0,0 @@
<?php
namespace app\queue\redis;
use support\exception\BusinessException;
use Webman\RedisQueue\Consumer;
class Notice implements Consumer
{
// 要消费的队列名
public $queue = 'notice';
// 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
public function consume($data)
{
try {
// 应用消息
$service = new \app\common\service\MessageNotificationsService();
$service->pushNotice($data);
// 邮件消息
if (isset($data['email']) && $data['email'] === 1) {
$emails = $service->getUserEmailsById($data['recipient']);
foreach ($emails as $email) {
if (empty($email)) {
continue;
}
\app\common\utils\email\MailSend::send($email, $data['title'], $data['content']);
}
}
return true;
} catch (\Throwable $th) {
flg("消息推送异常：".$th->getFile()."|".$th->getLine()."|".$th->getMessage());
throw new BusinessException($th->getFile()."|".$th->getLine()."|".$th->getMessage());
}
}
// 消费失败回调
/*
$package = [
'id' => 1357277951, // 消息ID
'time' => 1709170510, // 消息时间
'delay' => 0, // 延迟时间
'attempts' => 2, // 消费次数
'queue' => 'send-mail', // 队列名
'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
'max_attempts' => 5, // 最大重试次数
'error' => '错误信息' // 错误信息
]
*/
public function onConsumeFailure(\Throwable $e, $package)
{
echo "consume failure\n";
echo $e->getMessage() . "\n";
// 无需反序列化
var_export($package);
}
}
app/queue/redis/PermAgt.php
-95
View File
@@ -1,95 +0,0 @@
<?php
namespace app\queue\redis;
use support\Db;
use support\exception\BusinessException;
use Webman\RedisQueue\Consumer;
class PermAgt implements Consumer
{
// 要消费的队列名
public $queue = 'perm-agt';
// 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
public function consume($data)
{
try {
$corpId = $data['corp_id'];
$userId = $data['user_id'];
if ($corpId == 2) {
$roles = [
[
'user_id' => $userId,
'role_id' => 1
],
[
'user_id' => $userId,
'role_id' => 2
],
];
$systems = [
[
'user_id' => $userId,
'system_id' => 1
],
[
'user_id' => $userId,
'system_id' => 2
],
[
'user_id' => $userId,
'system_id' => 3
],
];
} else {
$roles = [
'user_id' => $userId,
'role_id' => 2
];
$systems = [
[
'user_id' => $userId,
'system_id' => 2
],
[
'user_id' => $userId,
'system_id' => 3
],
];
}
Db::table('user_roles')->insert($roles);
Db::table('user_systems')->insert($systems);
return true;
} catch (\Throwable $th) {
throw new BusinessException($th->getFile()."|".$th->getLine()."|".$th->getMessage());
}
}
// 消费失败回调
/*
$package = [
'id' => 1357277951, // 消息ID
'time' => 1709170510, // 消息时间
'delay' => 0, // 延迟时间
'attempts' => 2, // 消费次数
'queue' => 'send-mail', // 队列名
'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
'max_attempts' => 5, // 最大重试次数
'error' => '错误信息' // 错误信息
]
*/
public function onConsumeFailure(\Throwable $e, $package)
{
echo "consume failure\n";
echo $e->getMessage() . "\n";
// 无需反序列化
var_export($package);
}
}
app/queue/redis/Server.php
-57
View File
@@ -1,57 +0,0 @@
<?php
namespace app\queue\redis;
use GuzzleHttp\Client;
use GuzzleHttp\Exception\RequestException;
use support\exception\BusinessException;
use Webman\RedisQueue\Consumer;
class Server implements Consumer
{
// 要消费的队列名
public $queue = 'server';
// 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
public function consume($data)
{
try {
$source = $data['source'] === 'development' ? '-- 开发环境 --' : '** 生产环境 **';
$webhookUrl = 'https://open.feishu.cn/open-apis/bot/v2/hook/806c2e9d-6ca2-4345-a459-69bb2651706b';
$message = "====== ======\n";
$message .= $data['content']."\n";
$message .= "时间：".date('Y-m-d H:i:s', time())."\n";
$message .= "来源：".$source."\n";
$message .= "====== ======";
sendFeishuTextMessage($webhookUrl, $message);
return true;
} catch (\Throwable $th) {
var_dump('4 ：'.$th->getFile()."|".$th->getLine()."|".$th->getMessage());
throw new BusinessException($th->getFile()."|".$th->getLine()."|".$th->getMessage());
}
}
// 消费失败回调
/*
$package = [
'id' => 1357277951, // 消息ID
'time' => 1709170510, // 消息时间
'delay' => 0, // 延迟时间
'attempts' => 2, // 消费次数
'queue' => 'send-mail', // 队列名
'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
'max_attempts' => 5, // 最大重试次数
'error' => '错误信息' // 错误信息
]
*/
public function onConsumeFailure(\Throwable $e, $package)
{
echo "consume failure\n";
echo $e->getMessage() . "\n";
// 无需反序列化
var_export($package);
}
}
app/queue/redis/StockOrderPdf.php
-51
View File
@@ -1,51 +0,0 @@
<?php
namespace app\queue\redis;
use support\exception\BusinessException;
use Webman\RedisQueue\Consumer;
class StockOrderPdf implements Consumer
{
// 要消费的队列名
public $queue = 'stock-order-pdf';
// 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
public function consume($data)
{
try {
$service = new \app\crm\service\StockOrdersService();
$service->createdPdfFile($data['id']);
return true;
} catch (\Throwable $th) {
var_dump('4 ：'.$th->getFile()."|".$th->getLine()."|".$th->getMessage());
throw new BusinessException($th->getFile()."|".$th->getLine()."|".$th->getMessage());
}
}
// 消费失败回调
/*
$package = [
'id' => 1357277951, // 消息ID
'time' => 1709170510, // 消息时间
'delay' => 0, // 延迟时间
'attempts' => 2, // 消费次数
'queue' => 'send-mail', // 队列名
'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
'max_attempts' => 5, // 最大重试次数
'error' => '错误信息' // 错误信息
]
*/
public function onConsumeFailure(\Throwable $e, $package)
{
echo "consume failure\n";
echo $e->getMessage() . "\n";
// 无需反序列化
var_export($package);
}
}
app/queue/redis/WordPressPush.php
+138
View File
@@ -0,0 +1,138 @@
<?php
namespace app\queue\redis;
use Webman\RedisQueue\Consumer;
use support\Log;
use app\common\model\WordpressSite;
use app\common\model\WordpressUsedCar;
use app\common\model\UsedCarDetail;
use app\common\service\WordpressSiteService;
class WordPressPush implements Consumer
{
// 要消费的队列名
public $queue = 'wordpress-push';
// 连接名
public $connection = 'default';
/**
* 消费 WordPress 推送任务
* @param array $data
* - car_id: 车辆ID
* - site_id: WordPress站点ID（可选，默认使用启用的站点）
*/
public function consume($data)
{
$carId = $data['car_id'] ?? null;
$siteId = $data['site_id'] ?? null;
if (empty($carId)) {
Log::error('WordPress推送：车辆ID为空', ['data' => $data]);
return;
}
try {
// 获取车辆信息
$car = UsedCarDetail::find($carId);
if (!$car) {
Log::warning('WordPress推送：车辆不存在', ['car_id' => $carId]);
return;
}
// 获取WordPress站点
if ($siteId) {
$site = WordpressSite::find($siteId);
} else {
$site = WordpressSite::where('status', 1)->first();
}
if (!$site) {
$msg = '没有启用的WordPress站点';
Log::warning('WordPress推送失败：' . $msg, ['car_id' => $carId]);
// 推送失败，更新车辆状态为5
$car->update(['status' => 5]);
return;
}
// 检查是否已推送过
$existing = WordpressUsedCar::where('wp_site_id', $site->id)
->where('used_car_id', $carId)
->where('status', 1)
->first();
if ($existing) {
Log::info('WordPress推送：车辆已推送过，跳过', [
'car_id' => $carId,
'wp_product_id' => $existing->wp_product_id
]);
return;
}
// 执行推送
$wpService = new WordpressSiteService();
$result = $wpService->createProductForNewSite($site, $carId);
// 记录推送结果
if ($result['success']) {
// 推送成功，更新车辆状态为4
$car->update(['status' => 4]);
Log::info('WordPress推送成功，车辆状态已更新为推送成功', [
'car_id' => $carId,
'wp_product_id' => $result['product_id'] ?? null,
]);
} else {
// 推送失败，更新车辆状态为5
$car->update(['status' => 5]);
Log::error('WordPress推送失败，车辆状态已更新为推送失败', [
'car_id' => $carId,
'error' => $result['message'] ?? '未知错误',
]);
}
} catch (\Throwable $e) {
Log::error('WordPress推送异常', [
'car_id' => $carId,
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString(),
]);
// 推送失败，更新车辆状态为5
try {
if ($car) {
$car->update(['status' => 5]);
}
} catch (\Throwable $updateError) {
Log::error('更新车辆状态失败', [
'car_id' => $carId,
'error' => $updateError->getMessage(),
]);
}
throw $e;
}
}
/**
* 消费失败回调
*/
public function onConsumeFailure(\Throwable $e, $package)
{
Log::error('WordPress推送消费失败', [
'error' => $e->getMessage(),
'package' => $package,
]);
// 获取车辆ID并更新状态
$carId = $package['data']['car_id'] ?? null;
if ($carId) {
try {
UsedCarDetail::where('id', $carId)->update(['status' => 5]);
} catch (\Throwable $updateError) {
Log::error('推送失败后更新车辆状态异常', [
'car_id' => $carId,
'error' => $updateError->getMessage(),
]);
}
}
}
}
app/queue/redis/WsMsg.php
-68
View File
@@ -1,68 +0,0 @@
<?php
namespace app\queue\redis;
use app\common\model\UserSystems;
use Webman\RedisQueue\Consumer;
class WsMsg implements Consumer
{
// 要消费的队列名
public $queue = 'ws-msg';
// 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
public function consume($data)
{
try {
$ly = [
1 => 186,
3 => 226,
];
if (isset($data['is_first']) && $data['is_first'] === true) {
$num = (new \app\common\utils\approval\service\ApprovalFlowService())->getInitTodoCount($data['user_id']);
} else {
$num = $data['num'];
}
if ($data['event'] === 'approval.todo.counts') {
$systemIds = UserSystems::where('user_id', $data['user_id'])->pluck('system_id')->toArray();
if (count($systemIds) > 0) {
foreach ($systemIds as $systemId) {
if (isset($ly[$systemId])) {
$chain = \app\common\service\MenuBadgeService::buildBadgeChain(menuId: $ly[$systemId], leafCount: $num);
(new \app\common\utils\WsEvents())->pushWebsocket('approval.todo.counts', $data['user_id'], $chain);
}
}
}
}
} catch (\Throwable $th) {
dpr($th);
throw $th;
}
}
// 消费失败回调
/*
$package = [
'id' => 1357277951, // 消息ID
'time' => 1709170510, // 消息时间
'delay' => 0, // 延迟时间
'attempts' => 2, // 消费次数
'queue' => 'send-mail', // 队列名
'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
'max_attempts' => 5, // 最大重试次数
'error' => '错误信息' // 错误信息
]
*/
public function onConsumeFailure(\Throwable $e, $package)
{
echo "consume failure\n";
echo $e->getMessage() . "\n";
// 无需反序列化
var_export($package);
}
}
config/plugin/webman/redis-queue/process.php
+8 -9
View File
@@ -1,12 +1,11 @@
<?php
return [
// 'consumer' => [
// 'handler' => Webman\RedisQueue\Process\Consumer::class,
// 'count' => getenv('REDIS_QUEUE_PROCESS_COUNT'), // 可以设置多进程同时消费
// 'constructor' => [
// // 消费者类目录
// 'consumer_dir' => app_path() . '/queue/redis',
// 'consumer_dir' => app_path() . getenv('REDIS_QUEUE_CONSUMER_DIR')
// ]
// ]
'consumer' => [
'handler' => Webman\RedisQueue\Process\Consumer::class,
'count' => 1, // 消费者进程数量
'constructor' => [
// 消费者类目录
'consumer_dir' => app_path() . '/queue/redis',
]
]
];