czg_team/p_ysk
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

营销短信查询统计记录

Browse Source
This commit is contained in:
ASUS
2025-11-11 10:10:09 +08:00
parent 29475b7da4
commit e88b220650
5 changed files with 32 additions and 450 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
app/command/ApplySmsTemp.php
View File

@@ -1,105 +0,0 @@
<?php
namespace app\command;
use app\common\model\RabbitMqConfig;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use support\Log;
use Webman\RedisQueue\Redis;
// 申请新短信模版/营销短信发送
class ApplySmsTemp extends Command
{
protected static $defaultName = 'applysmstemp';
protected static $defaultDescription = 'applysmstemp';
/**
* @return void
*/
protected function configure()
{
$this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}
/**
* @param InputInterface $input
* @param OutputInterface $output
* @return int
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
// $str = '亲爱的${用户昵称}您好，${店铺名称}祝您生日快乐！感谢您一直的陪伴。为您准备了${数量}张超值优惠券，已放入账户，愿我们的礼物能为您增添一份喜悦。拒收请回复R';
// $templateContent_arr = ShopInfo::handleSmsTemplates($str);
// $a = '{"店铺名称":"火锅店","用户昵称":"嘿嘿","数量":"123"}';
// p(replace_json_keys($a, json_encode($templateContent_arr, 256), []));
// $t_str = replace_placeholder_keys($str, $templateContent_arr);
// p($templateContent_arr, $t_str);
// $stortime = strtotime('2025-10-16 19:03:00');
// \support\Redis::setEx('expired:sms:9995', $stortime - time(), 1);
// Log::info('定时发送，已存入redis');
// p(123123);
$host = config('cons.mq.host');
$port = config('cons.mq.port');
$user = config('cons.mq.user');
$password = config('cons.mq.password');
$queue = config('cons.mq.queue_t') . '-apply.sms.temp';
// 防止空闲时间断线必须设置心跳
$connection = new AMQPStreamConnection($host, $port, $user, $password,
'/',
false,
'AMQPLAIN',
null,
'en_US',
60,
60,
null,
false,
30
);
$rabbit_channel = $connection->channel();
$rabbit_channel->queue_declare($queue, false, true, false, false, false, [
// 'x-message-ttl' => ['I', 180000]
]);
$callback = function ($msg) use ($queue){
$data = $msg->body;
Log::info('MQ收到消息[申请新短信模版/营销短信发送/微信模版消息]--->' . $data . "\n");
$arr = explode(',', $data);
if(isset($arr[2])) {
$type = $arr[2];
if($type == 'applySmsTemp') {
// 发给队列 短信模版审核
$a = Redis::send('apply.sms.temp', $data);
Log::info('消息队列投递结果----》' . $a);
}elseif ($type == 'sendMarkSms') {
// 发给队列 营销短信发送
Redis::send('send.mark.sms', $data);
}elseif ($type == 'sendWechatTemp') {
// 发给队列 发送微信模版消息
Redis::send('send.wechat.temp', $data);
}
}else {
Log::info('MQ收到消息[申请新短信模版/营销短信发送/微信模版消息]格式错误');
}
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$rabbit_channel->basic_consume($queue, '', false, false, false, false, $callback);
while ($rabbit_channel->is_consuming()) {
$rabbit_channel->wait();
}
$rabbit_channel->close();
$connection->close();
return self::SUCCESS;
}
}

82
app/command/BirthdayGiftSms.php
View File

@@ -1,82 +0,0 @@
<?php
namespace app\command;
use app\common\model\MqLog;
use app\common\model\OrderInfo;
use app\common\model\RabbitMqConfig;
use ba\Random;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use support\think\Db;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use support\Log;
use Webman\RedisQueue\Redis;
// 生日有礼
class BirthdayGiftSms extends Command
{
protected static $defaultName = 'birthdaygiftsms';
protected static $defaultDescription = 'birthdaygiftsms';
/**
* @return void
*/
protected function configure()
{
$this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}
/**
* @param InputInterface $input
* @param OutputInterface $output
* @return int
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
$host = config('cons.mq.host');
$port = config('cons.mq.port');
$user = config('cons.mq.user');
$password = config('cons.mq.password');
$queue = config('cons.mq.queue_t') . '-birthday.gift.sms';
// 防止空闲时间断线必须设置心跳
$connection = new AMQPStreamConnection($host, $port, $user, $password,
'/',
false,
'AMQPLAIN',
null,
'en_US',
60,
60,
null,
false,
30
);
$rabbit_channel = $connection->channel();
$rabbit_channel->queue_declare($queue, false, true, false, false, false, [
// 'x-message-ttl' => ['I', 180000]
]);
$callback = function ($msg) use ($queue){
$date_time = date('Y-m-d H:i:s');
$data = $msg->body;
Log::info('MQ收到消息[生日有礼]--->' . $data . '--->' . $date_time . "\n");
// 发给队列
Redis::send('birthday.gift.sms', $data);
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$rabbit_channel->basic_consume($queue, '', false, false, false, false, $callback);
while ($rabbit_channel->is_consuming()) {
$rabbit_channel->wait();
}
$rabbit_channel->close();
$connection->close();
return self::SUCCESS;
}
}

38
app/command/QuerySmsStatus.php
View File

@@ -91,7 +91,7 @@ class QuerySmsStatus extends Command
}
// 营销短信
// 营销短信发送结果查询
$bir_list = Db::table('sms_push_event_record')->where(['status' => 1])->select()->toArray();
if($bir_list) {
Log::info('营销短信发送状态查询开始->' . json_encode($bir_list));
@@ -122,10 +122,6 @@ class QuerySmsStatus extends Command
'error' => $push_resp,
];
$update_status = Db::table('sms_push_event_record')->where(['id' => $value['id']])->update($update_data);
// 只要有一个发送成功的就更改状态发送成功
if($push_status == 2) {
Db::table('sms_push_event')->where(['id' => $value['event_id']])->update(['status' => 2]);
}
Log::info('营销短信查询完毕数据库更新结果---》[ID]' . $value['id'] . '更新结果-->' . $update_status . '-->更新数据 . ' . json_encode($update_data));
}
Db::commit();
@@ -133,12 +129,42 @@ class QuerySmsStatus extends Command
Db::rollback();
Log::info('短信查询错误---》[ID]' . $e->getMessage());
}
sleep(1);
}
}else {
Log::info('没有营销短信状态查询的数据');
}
// 营销短信发送失败成功统计
$sms_push_event_list = Db::name('sms_push_event')->where(['status' => 1])->select()->toArray();
if($sms_push_event_list) {
Log::info('营销短信发送失败成功统计开始->' . json_encode($sms_push_event_list));
foreach ($sms_push_event_list as $k => $v) {
$result = Db::table('sms_push_event_record')
->where('event_id', $v['id']) // 统一筛选 event_id=2 的数据
->field([
// 1. event_id=2 的实际人数
'COUNT(*) AS total_count',
// 成功人数
'COUNT(CASE WHEN status = 1 THEN 1 END) AS success_count',
// 失败人数
'COUNT(CASE WHEN status = -1 THEN 1 END) AS fail_count'
])
->find(); // 单条结果（返回关联数组）
// 如果成功人数和失败人数等于实际人数，代表发送任务已经完成
if(($result['success_count'] + $result['fail_count']) >= $result['total_count']) {
Log::info('发送任务【' .$v['id']. '】已经完成-> 成功' . $result['success_count'] . ' 失败' . $result['fail_count']);
$update = [
'status' => 2,
'actual_num' => $result['total_count'],
'success_num' => $result['success_count'],
];
Db::name('sms_push_event')->where(['id' => $v['id']])->update($update);
}
}
}else {
Log::info('没有营销短信发送统计数据');
}
return self::SUCCESS;
}

138
app/command/RabbOrderPrint.php
View File

@@ -1,138 +0,0 @@
<?php
namespace app\command;
use app\common\model\MqLog;
use app\common\model\OrderInfo;
use app\common\model\RabbitMqConfig;
use ba\Random;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use support\think\Db;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use support\Log;
class RabbOrderPrint extends Command
{
protected static $defaultName = 'rabborderprint';
protected static $defaultDescription = 'rabborderprint';
/**
* @return void
*/
protected function configure()
{
$this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}
/**
* @param InputInterface $input
* @param OutputInterface $output
* @return int
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
$host = config('cons.mq.host');
$port = config('cons.mq.port');
$user = config('cons.mq.user');
$password = config('cons.mq.password');
$queue = config('cons.mq.queue_t') . '-order.print.queue';
// 防止空闲时间断线必须设置心跳
$connection = new AMQPStreamConnection($host, $port, $user, $password,
'/',
false,
'AMQPLAIN',
null,
'en_US',
60,
60,
null,
false,
30
);
$rabbit_channel = $connection->channel();
$rabbit_channel->queue_declare($queue, false, true, false, false, false, [
'x-message-ttl' => ['I', 180000]
]);
$callback = function ($msg) use ($queue){
$date_time = date('Y-m-d H:i:s');
$data = $msg->body;
Log::info('MQ收到消息[订单打印]--->' . $data . '--->' . $date_time . "\n");
// 如果是数字。则打印订单
$is_log = false;
$curl_error = '';
print_r($data . "\r\n");
$order_id_t = strpos($data, '_');
print_r($order_id_t . "嘿嘿\r\n");
if($order_id_t !== false) {
print_r($order_id_t . "\r\n");
$order_id = substr($data, 0, $order_id_t);
}else {
print_r( "无下划线\r\n");
$order_id = $data;
}
print_r($order_id . "\r\n");
$order = Db::table('tb_order_info')->where(['id' => $order_id])->find();
if($order) {
$rand = 'cashier_order' . Random::build();
$send_id = 'cashier_'. $order['shop_id'];
// 收银机打印订单
$snd_data = [
'msg' => '收银机订单打印',
'type' => 'cashier',
'operate_type' => 'order_print',
'data_type' => 'order',
'status' => 1,
'method' => 'sendToUid',
'send_num' => 0,
'send_id' => $send_id,
'msg_id' => $rand,
'data' => $data
];
$snd_data_json = json_encode($snd_data);
// 内部给收银机推送打订单
$url = 'http://127.0.0.1:8686/?method=sendToUid&account='. $send_id .'&params=' . $snd_data_json; // 替换为你的服务地址和端口
$ch = curl_init();
curl_setopt($ch, CURLOPT_URL, $url);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
$output = curl_exec($ch);
if (curl_errno($ch)) {
$is_log = true;
$curl_error = 'Curl error: ' . curl_error($ch);
}
curl_close($ch);
Log::info('内部通讯结果(订单打印)--->' . $output . '通讯ID--->cashier_' .$order['shop_id']. '---' . date('Y-m-d H:i:s') . "\n");
}else {
$curl_error = 'order Data is empty';
$is_log = true;
}
if($is_log) {
Db::table('tb_mq_log')->insert([
'queue' => $queue,
'msg' => $data,
'type' => 'orderPrint',
'plat' => 'Cashier',
'create_time' => date('Y-m-d H:i:s'),
'fail_time' => date('Y-m-d H:i:s'),
'err_info' => $curl_error,
]);
}
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$rabbit_channel->basic_consume($queue, '', false, false, false, false, $callback);
while ($rabbit_channel->is_consuming()) {
$rabbit_channel->wait();
}
$rabbit_channel->close();
$connection->close();
return self::SUCCESS;
}
}

119
app/command/RabbProductUpdate.php
View File

@@ -1,119 +0,0 @@
<?php
namespace app\command;
use app\common\model\MqLog;
use app\common\model\RabbitMqConfig;
use ba\Random;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use support\think\Db;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Output\OutputInterface;
use support\Log;
class RabbProductUpdate extends Command
{
protected static $defaultName = 'rabbproductupdate';
protected static $defaultDescription = 'rabbproductupdate';
/**
* @return void
*/
protected function configure()
{
$this->addArgument('name', InputArgument::OPTIONAL, 'Name description');
}
/**
* @param InputInterface $input
* @param OutputInterface $output
* @return int
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
$host = config('cons.mq.host');
$port = config('cons.mq.port');
$user = config('cons.mq.user');
$password = config('cons.mq.password');
$queue = config('cons.mq.queue_t') . '-product.info.change.queue';
// 防止空闲时间断线必须设置心跳
$connection = new AMQPStreamConnection($host, $port, $user, $password,
'/',
false,
'AMQPLAIN',
null,
'en_US',
60,
60,
null,
false,
30
);
$rabbit_channel = $connection->channel();
$rabbit_channel->queue_declare($queue, false, true, false, false, false, [
// 'x-message-ttl' => ['I', 180000]
]);
$callback = function ($msg) use ($queue){
$date_time = date('Y-m-d H:i:s');
$data = $msg->body;
Log::info('MQ收到消息[商品更新]--->' . $data . '--->' . $date_time . "\n");
// 如果是数字。则打印订单
$is_log = false;
$curl_error = '';
if(is_numeric($data)) {
$rand = 'product_update' . Random::build();
$send_id = 'product_update'. $data;
$method = 'sendToGroup';
$snd_data = [
'msg' => '商品更新',
'type' => 'product',
'operate_type' => 'product_update',
'data_type' => 'product_update',
'status' => 1,
'method' => $method,
'send_num' => 0,
'send_id' => $send_id,
'msg_id' => $rand,
'data' => $data
];
$snd_data_json = json_encode($snd_data);
$url = 'http://127.0.0.1:8686/?method=' . $method . '&account='. $send_id .'&params=' . $snd_data_json; // 替换为你的服务地址和端口
$ch = curl_init();
curl_setopt($ch, CURLOPT_URL, $url);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
$output = curl_exec($ch);
if (curl_errno($ch)) {
$is_log = true;
$curl_error = 'Curl error: ' . curl_error($ch);
}
curl_close($ch);
Log::info('内部通讯结果(商品更新)--->' . $output . '通讯ID--->shop_id-->' .$send_id. '---' . date('Y-m-d H:i:s') . "\n");
if($is_log) {
Db::table('tb_mq_log')->insert([
'queue' => $queue,
'msg' => $data,
'type' => 'product_update',
'plat' => 'product',
'create_time' => date('Y-m-d H:i:s'),
'fail_time' => date('Y-m-d H:i:s'),
'err_info' => $curl_error,
]);
}
}
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$rabbit_channel->basic_consume($queue, '', false, false, false, false, $callback);
while ($rabbit_channel->is_consuming()) {
$rabbit_channel->wait();
}
$rabbit_channel->close();
$connection->close();
return self::SUCCESS;
}
}