This commit is contained in:
2026-01-05 14:33:21 +08:00
commit ee348b4925
451 changed files with 39954 additions and 0 deletions
+83
View File
@@ -0,0 +1,83 @@
<?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;
}
}
+72
View File
@@ -0,0 +1,72 @@
<?php
namespace app\queue\redis;
use support\Db;
use support\Log;
use Webman\RedisQueue\Consumer;
class AuditLog implements Consumer
{
// 要消费的队列名
public $queue = 'audit_log';
// 连接名,对应 plugin/webman/redis-queue/redis.php 里的连接`
public $connection = 'default';
// 消费
public function consume($data)
{
try {
$now = (int)($data['ts'] ?? time());
$row = [
// 'trace_id' => $data['trace_id'] ?? null,
'user_id' => $data['user_id'] ?? null,
'user_name' => $data['user_name'] ?? null,
'route' => $data['route'] ?? null,
'method' => $data['method'] ?? null,
'ip' => $data['ip'] ?? null,
'ua' => $data['ua'] ?? null,
'action' => $data['action'] ?? null,
'status' => (int)($data['status'] ?? 1),
'http_code' => (int)($data['http_code'] ?? 200),
'latency_ms' => $data['latency_ms'] ?? null,
'params_brief' => !empty($data['params_brief']) ? json_encode($data['params_brief'], JSON_UNESCAPED_UNICODE) : null,
'changes' => !empty($data['changes']) ? json_encode($data['changes'], JSON_UNESCAPED_UNICODE) : null,
'message' => $data['message'] ?? null,
'created_at' => $now,
'updated_at' => $now,
];
Db::table('sys_op_log')->insert($row);
} catch (\Throwable $e) {
Log::error('audit_log consume error: '.$e->getMessage());
return false; // 允许重试与失败处理
}
return true;
}
// 消费失败回调
/*
$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);
Log::error('audit_log message failed after retries: '.json_encode($package, JSON_UNESCAPED_UNICODE));
}
}
+51
View File
@@ -0,0 +1,51 @@
<?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);
}
}
+131
View File
@@ -0,0 +1,131 @@
<?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);
}
}
+222
View File
@@ -0,0 +1,222 @@
<?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);
}
}
+207
View File
@@ -0,0 +1,207 @@
<?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_bak implements Consumer
{
// 要消费的队列名
public $queue = 'inquiry-to-crm-new_bak';
// 连接名,对应 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 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);
} 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);
} else {
// $notice = '重新分配';
$title = '[询单·分配] 新的客户线索【' . $info->order_no . '】,请及时处理';
$content = '[询单·分配] 新的客户线索【' . $info->order_no . '】,请及时处理';
$recipient = [$info->recipient];
$this->pushWs($title, $content, $recipient);
}
}
$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);
}
if ($info->status == 2) {
// $notice = '分配';
$title = '[询单·分配] 新的客户线索【' . $info->order_no . '】,请及时处理';
$content = '[询单·分配] 新的客户线索【' . $info->order_no . '】,请及时处理';
$recipient = [$info->recipient];
$this->pushWs($title, $content, $recipient);
}
}
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);
}
}
+61
View File
@@ -0,0 +1,61 @@
<?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);
}
}
+95
View File
@@ -0,0 +1,95 @@
<?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);
}
}
+57
View File
@@ -0,0 +1,57 @@
<?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);
}
}
+51
View File
@@ -0,0 +1,51 @@
<?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);
}
}
+68
View File
@@ -0,0 +1,68 @@
<?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);
}
}
+21
View File
@@ -0,0 +1,21 @@
<?php
namespace app\queue\utils;
class Translat
{
/**
* 从$data中提取出源文本字段,并从$data里删除它们
*/
public static function extractSourceFields(array &$data,array $translat): array
{
$source = [];
foreach ($translat as $field) {
if (array_key_exists($field, $data)) {
$source[$field] = $data[$field];
unset($data[$field]);
}
}
return $source;
}
}
+138
View File
@@ -0,0 +1,138 @@
<?php
namespace app\queue\utils;
use support\Log;
class XunfeiFanyi
{
protected $error = "";
public function tocurl($url, $header, $content)
{
$ch = curl_init();
if(substr($url,0,5)=='https'){
curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, false); // 跳过证书检查
curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, false); // 从证书中检查SSL加密算法是否存在
curl_setopt($ch, CURLOPT_SSLVERSION, 1);
}
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
curl_setopt($ch, CURLOPT_URL, $url);
if (is_array($header)) {
curl_setopt($ch, CURLOPT_HTTPHEADER, $header);
}
curl_setopt($ch, CURLOPT_POST, true);
if (!empty($content)) {
if (is_array($content)) {
curl_setopt($ch, CURLOPT_POSTFIELDS, http_build_query($content));
} else if (is_string($content)) {
curl_setopt($ch, CURLOPT_POSTFIELDS, $content);
}
}
$response = curl_exec($ch);
$error=curl_error($ch);
//var_dump($error);
if($error){
die($error);
}
$header = curl_getinfo($ch);
curl_close($ch);
$data = array('header' => $header,'body' => $response);
return $data;
}
public function xfyun($text = "中华人民共和国于1949年成立", $from = "cn", $to = "en")
{
//在控制台-我的应用-机器翻译获取
$app_id = "3d46198c";
//在控制台-我的应用-机器翻译获取
$api_sec = "ZWY2MmUyNjJjYmFkOTFlYTVmYWVhMGYy";
//在控制台-我的应用-机器翻译获取
$api_key = "b6d805f240176ecaea85130cb5f2bca1";
// 机器翻译接口地址
$url = "https://ntrans.xfyun.cn/v2/ots";
//body组装
// $text = "中华人民共和国于1949年成立";//待翻译文本
$body = json_encode($this->getBody($app_id, $from, $to, $text));
// 组装http请求头
$date =gmdate('D, d M Y H:i:s') . ' GMT';
$digestBase64 = "SHA-256=".base64_encode(hash("sha256", $body, true));
$builder = sprintf("host: %s
date: %s
POST /v2/ots HTTP/1.1
digest: %s", "ntrans.xfyun.cn", $date, $digestBase64);
// echo($builder);
$sha = base64_encode(hash_hmac("sha256", $builder, $api_sec, true));
$authorization = sprintf("api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"", $api_key,"hmac-sha256",
"host date request-line digest", $sha);
$header = [
"Authorization: ".$authorization,
'Content-Type: application/json',
'Accept: application/json,version=1.0',
'Host: ntrans.xfyun.cn',
'Date: ' .$date,
'Digest: '.$digestBase64
];
$response = $this->tocurl($url, $header, $body);
// var_dump($response['body']);
return $response;
}
public function getBody($app_id, $from, $to, $text)
{
$common_param = [
'app_id' => $app_id
];
$business = [
'from' => $from,
'to' => $to,
];
$data = [
"text" => base64_encode($text)
];
return $body = [
'common' => $common_param,
'business' => $business,
'data' => $data
];
}
// 翻译
public function fanyi($text = "", $from = "", $to = "")
{
try {
$res = $this->xfyun($text, $from, $to);
$data = json_decode($res["body"], true);
if ($data["code"] === 0) {
return $data["data"]["result"];
} else {
$this->error = $data["message"];
Log::error("XunfeiFanyi 失败: ".$data["message"],[$data]);
return false;
}
} catch (\Throwable $th) {
$this->error = $th->getMessage();
Log::error("XunfeiFanyi 错误: ".$th->getMessage());
return false;
}
}
// 异常
public function error()
{
return $this->error;
}
}