444 lines
23 KiB
PHP
444 lines
23 KiB
PHP
<?php
|
|
namespace app\console\command;
|
|
|
|
use app\common\service\QueueService;
|
|
use app\console\model\OrderGoods;
|
|
use think\console\Command;
|
|
use think\console\Input;
|
|
use think\console\Output;
|
|
use app\store\model\order\Record as RecordModel;
|
|
use app\common\service\SybOil;
|
|
use think\facade\Db;
|
|
|
|
class CallbackQueue extends Command
|
|
{
|
|
protected $queueService;
|
|
|
|
protected function configure()
|
|
{
|
|
$this->setName('callback:queue')
|
|
->setDescription('Process messages from the Redis queue');
|
|
}
|
|
|
|
protected function execute(Input $input, Output $output)
|
|
{
|
|
$this->queueService = new QueueService();
|
|
|
|
while (true) {
|
|
$message = $this->queueService->pop('callback_queue');
|
|
if ($message) {
|
|
// 处理消息
|
|
$this->processMessage($message);
|
|
} else {
|
|
// 暂停一段时间再检查队列
|
|
sleep(1);
|
|
}
|
|
}
|
|
}
|
|
|
|
// protected function processMessage($message)
|
|
// {
|
|
// $SybOil = new SybOil();
|
|
// $SybOil->saveLogStore('callback_queue队列数据处理：【'.$message.'】');
|
|
// // 拿到队列数据处理
|
|
// $data = json_decode($message, true);
|
|
// // 检查解码是否成功
|
|
// if ($data === false) {
|
|
// $SybOil->saveLogStore('callback_queue队列数据解码失败：' . json_last_error_msg());
|
|
// return;
|
|
// }
|
|
// // 确保 $data 中包含 header 和 res
|
|
// if (!isset($data['header']) || !isset($data['res'])) {
|
|
// $SybOil->saveLogStore('数据格式不正确：【' . $message . '】');
|
|
// return;
|
|
// }
|
|
// $headers = $data['header'];
|
|
// $res = $data['res'];
|
|
// // 此处可以根据业务逻辑处理接收到的数据
|
|
// try {
|
|
// switch ($headers['method']) {
|
|
// case "syb.notify.pay.result":
|
|
// // 二维码交易结果通知
|
|
// if($res['oilCardType'] == 'card') {
|
|
// $where = array(
|
|
// ['oilCardNo', '=', $res['oilCardNo']],
|
|
// ['goods_price', '=', $res['amount']],
|
|
// );
|
|
// }else{
|
|
// $where = array(
|
|
// ['shareNo', '=', $res['shareNo']],
|
|
// ['goods_price', '=', $res['amount']],
|
|
// );
|
|
// }
|
|
// $orderGoods = Db::name('order_goods')
|
|
// ->where($where)
|
|
// ->find();
|
|
// if($orderGoods['order_goods_id']) {
|
|
// $field['qrCode'] = $res['qrCode'];
|
|
// if ($res['status'] == 'success') {
|
|
// $field['use_status'] = 1;
|
|
// } else {
|
|
// $field['use_text'] = $res['errMessage'];
|
|
// }
|
|
// // 更新用户使用状态
|
|
// Db::name('order_goods')
|
|
// ->where('order_goods_id', $orderGoods['order_goods_id'])
|
|
// ->update($field);
|
|
// }else{
|
|
// $SybOil->saveLogStore('未找到订单信息：【'.$message.'】');
|
|
// }
|
|
// break;
|
|
// case "syb.notify.oilCard.change":
|
|
// // 油卡变动通知
|
|
// $where = array(
|
|
// ['oilCardNo', '=', $res['oilCardNo']],
|
|
// ['goods_price', '=', $res['amount']],
|
|
// );
|
|
// $orderGoods = Db::name('order_goods')
|
|
// ->where($where)
|
|
// ->find();
|
|
// $res['order_id'] = $orderGoods['order_id'];
|
|
// $res['order_goods_id'] = $orderGoods['order_goods_id'];
|
|
// $res['user_id'] = $orderGoods['user_id'];
|
|
// $res['store_id'] = $orderGoods['store_id'];
|
|
// $RecordModel = new RecordModel();
|
|
// // 油卡变动记录写入结果
|
|
// $RecordModel->add($res);
|
|
// // 更新油券使用状态
|
|
// if($orderGoods['order_goods_id'] && $res['optType'] == 'add_oil'){
|
|
// Db::name('order_goods')
|
|
// ->where('order_goods_id', $orderGoods['order_goods_id'])
|
|
// ->update(array(
|
|
// 'use_status' => 1,
|
|
// 'use_time' => time(),
|
|
// 'use_text' => '加油',
|
|
// ));
|
|
// // 修改主订单为已完成
|
|
// Db::name('order')
|
|
// ->where('order_id', $orderGoods['order_id'])
|
|
// ->update(array(
|
|
// 'order_status' => 30
|
|
// ));
|
|
// // 更新用户使用状态
|
|
// Db::name('account')
|
|
// ->where('account_id', $orderGoods['account_id'])
|
|
// ->update(['last_called_at' => time(), 'state' => 0]);
|
|
// $SybOil->saveLogStore('加油回调处理成功了：'.json_encode($res));
|
|
// }else{
|
|
// $SybOil->saveLogStore('没有处理结果：【'.$message.'】');
|
|
// }
|
|
// break;
|
|
// case "syb.notify.oilCard.share":
|
|
// // 分油结果通知
|
|
// if($res['oilCardType'] == 'card') {
|
|
// $where = array(
|
|
// ['oilCardNo', '=', $res['oilCardNo']],
|
|
// ['goods_price', '=', $res['amount']],
|
|
// );
|
|
// }else{
|
|
// $where = array(
|
|
// ['shareNo', '=', $res['shareNo']],
|
|
// ['goods_price', '=', $res['amount']],
|
|
// );
|
|
// }
|
|
// $orderGoods = Db::name('order_goods')
|
|
// ->where($where)
|
|
// ->find();
|
|
// if($orderGoods['order_goods_id']) {
|
|
// $field = ['oil_time' => time()];
|
|
// if ($res['status'] == 'success') {
|
|
// $field['oil_status'] = 1;
|
|
// } else {
|
|
// $field['oil_status'] = 0;
|
|
// $field['oil_text'] = $res['msg'];
|
|
// }
|
|
// Db::name('order_goods')
|
|
// ->where('order_goods_id', $orderGoods['order_goods_id'])
|
|
// ->update($field);
|
|
//
|
|
// $SybOil->saveLogStore('分油回调处理成功了：'.json_encode($res));
|
|
// }else{
|
|
// $this->queueService->push('callback_queue', json_encode(['header'=>$headers,'res'=>$res],320));
|
|
// $SybOil->saveLogStore('重新加入队列：【'.$message.'】');
|
|
// }
|
|
// break;
|
|
// default:
|
|
// $SybOil->saveLogStore('队列数据处理异常，未知：【'.$message.'】');
|
|
// }
|
|
// } catch (\Exception $e) {
|
|
// $SybOil->saveLogStore(
|
|
// '队列回调处理异常：' . $e->getMessage() .
|
|
// ' 在文件：' . $e->getFile() .
|
|
// ' 第：' . $e->getLine() .
|
|
// ' 行。堆栈跟踪：' . $e->getTraceAsString()
|
|
// );
|
|
//
|
|
// }
|
|
// }
|
|
|
|
protected function processMessage($message)
|
|
{
|
|
$SybOil = new SybOil();
|
|
$SybOil->saveLogStore('callback_queue队列数据处理：【'.$message.'】');
|
|
// 拿到队列数据处理
|
|
$data = json_decode($message, true);
|
|
// 检查解码是否成功
|
|
if ($data === false) {
|
|
$SybOil->saveLogStore('callback_queue队列数据解码失败：' . json_last_error_msg());
|
|
return;
|
|
}
|
|
// 确保 $data 中包含 header 和 res
|
|
if (!isset($data['header']) || !isset($data['res'])) {
|
|
$SybOil->saveLogStore('数据格式不正确：【' . $message . '】');
|
|
return;
|
|
}
|
|
$headers = $data['header'];
|
|
$res = $data['res'];
|
|
// 此处可以根据业务逻辑处理接收到的数据
|
|
try {
|
|
switch ($headers['method']) {
|
|
case "use":
|
|
// 油卡变动通知
|
|
$where = array(
|
|
['shareNo', '=', $res['trade_no']],
|
|
);
|
|
$orderGoods = Db::name('order_goods')
|
|
->where($where)
|
|
->find();
|
|
$res['order_id'] = $orderGoods['order_id'];
|
|
$res['order_goods_id'] = $orderGoods['order_goods_id'];
|
|
$res['user_id'] = $orderGoods['user_id'];
|
|
$res['store_id'] = $orderGoods['store_id'];
|
|
$res['optType'] = 'add_oil';
|
|
$RecordModel = new RecordModel();
|
|
// 油卡变动记录写入结果
|
|
$RecordModel->add($res);
|
|
// 更新油券使用状态
|
|
if($orderGoods['order_goods_id'] && $res['useStatus'] == '2'){
|
|
Db::name('order_goods')
|
|
->where('order_goods_id', $orderGoods['order_goods_id'])
|
|
->update(array(
|
|
'use_status' => 1,
|
|
'use_time' => time(),
|
|
'use_text' => '加油',
|
|
));
|
|
// 订单下的券都已使用
|
|
if(Db::name('order_goods')->where([['order_id' , '=', $orderGoods['order_id']], ['use_status', '=', '0']])->count() == 0) {
|
|
// 修改主订单为已完成
|
|
Db::name('order')
|
|
->where('order_id', $orderGoods['order_id'])
|
|
->update(array(
|
|
'order_status' => 30
|
|
));
|
|
}
|
|
// 更新用户使用状态
|
|
Db::name('account')
|
|
->where('account_id', $orderGoods['account_id'])
|
|
->update(['last_called_at' => time(), 'state' => 0]);
|
|
$SybOil->saveLogStore('加油回调处理成功了：'.json_encode($res));
|
|
}else{
|
|
$SybOil->saveLogStore('没有处理结果：【'.$message.'】');
|
|
}
|
|
break;
|
|
case "oil":
|
|
// 分油结果通知
|
|
$where = array(
|
|
['shareNo', '=', $res['trade_no']],
|
|
);
|
|
$orderGoods = Db::name('order_goods')
|
|
->where($where)
|
|
->find();
|
|
if($orderGoods['order_goods_id'] && $res['useStatus'] == '1') {
|
|
$field = ['oil_time' => time()];
|
|
$field['oil_status'] = 1;
|
|
$field['qrCode'] = $res['useUrl'] ?? '';
|
|
Db::name('order_goods')
|
|
->where('order_goods_id', $orderGoods['order_goods_id'])
|
|
->update($field);
|
|
|
|
$SybOil->saveLogStore('分油回调处理成功了：'.json_encode($res));
|
|
}else{
|
|
// $this->queueService->push('callback_queue', json_encode(['header'=>$headers,'res'=>$res],320));
|
|
// $SybOil->saveLogStore('重新加入队列：【'.$message.'】');
|
|
}
|
|
break;
|
|
case "use_partners":
|
|
// 油卡变动通知
|
|
$where = array(
|
|
['shareNo', '=', $res['trade_no']],
|
|
);
|
|
$order = Db::name('partners_order')
|
|
->where($where)
|
|
->find();
|
|
$res['order_id'] = $order['order_id'];
|
|
$res['order_goods_id'] = $order['goods_id'];
|
|
$res['user_id'] = 0;
|
|
$res['store_id'] = $order['store_id'];
|
|
$res['optType'] = 'add_oil';
|
|
$RecordModel = new RecordModel();
|
|
// 油卡变动记录写入结果
|
|
$RecordModel->add($res);
|
|
// 更新油券使用状态
|
|
if($order['order_id'] && $res['useStatus'] == '2'){
|
|
Db::name('partners_order')
|
|
->where('order_id', $order['order_id'])
|
|
->update(array(
|
|
'use_status' => 1,
|
|
'use_time' => time(),
|
|
'use_text' => '加油',
|
|
));
|
|
$SybOil->saveLogStore('加油回调处理成功了：'.json_encode($res));
|
|
|
|
// 通知对接公司
|
|
$queueService = new QueueService();
|
|
$queueService->push('notify_queue', json_encode([
|
|
'order_id' => $order['order_id'],
|
|
'optType' => 'add_oil',
|
|
'retry_count' => 0,
|
|
], 320));
|
|
}else{
|
|
$SybOil->saveLogStore('没有处理结果：【'.$message.'】');
|
|
}
|
|
break;
|
|
case "oil_partners":
|
|
// 分油结果通知
|
|
$where = array(
|
|
['shareNo', '=', $res['trade_no']],
|
|
);
|
|
$order = Db::name('partners_order')
|
|
->where($where)
|
|
->find();
|
|
if($order['order_id'] && $res['useStatus'] == '1') {
|
|
$field = ['oil_time' => time()];
|
|
$field['oil_status'] = 1;
|
|
$field['qrCode'] = $res['useUrl'] ?? '';
|
|
Db::name('partners_order')
|
|
->where('order_id', $order['order_id'])
|
|
->update($field);
|
|
|
|
$SybOil->saveLogStore('分油回调处理成功了：'.json_encode($res));
|
|
|
|
// 通知对接公司
|
|
$queueService = new QueueService();
|
|
$queueService->push('notify_queue', json_encode([
|
|
'order_id' => $order['order_id'],
|
|
'optType' => 'share_oil',
|
|
'retry_count' => 0,
|
|
], 320));
|
|
}else{
|
|
// $this->queueService->push('callback_queue', json_encode(['header'=>$headers,'res'=>$res],320));
|
|
// $SybOil->saveLogStore('重新加入队列：【'.$message.'】');
|
|
}
|
|
break;
|
|
case "qyt_consume":
|
|
// 企油通 7.10 运单加油明细推送（一次可能多条；企油通可能重推 -> 全流程幂等）
|
|
foreach ((array)($res['order_list'] ?? []) as $c) {
|
|
$code = $c['waybill_no_code'] ?? '';
|
|
$order = Db::name('partners_order')->where('waybill_no_code', $code)->find();
|
|
if (!$order) {
|
|
$SybOil->saveLogStore('企油通消费回调未匹配订单：' . json_encode($c, 320));
|
|
continue;
|
|
}
|
|
$payTime = !empty($c['pay_time']) ? (strtotime($c['pay_time']) ?: time()) : time();
|
|
// 一单可以分多次加油：每条推送是一次加油，consume_amount = 本次占用运单金额（分）
|
|
$consumeFen = (int)($c['consume_amount'] ?? 0);
|
|
$consumeYuan = number_format($consumeFen / 100, 2, '.', '');
|
|
$orderCode = trim((string)($c['order_code'] ?? ''));
|
|
// 去重键：文档里 order_code 是选填，缺失时用本笔消费的特征字段做指纹，
|
|
// 否则同一运单的第二笔加油会被唯一键误判为重复推送而丢掉（口径与迁移脚本回填一致）
|
|
$consumeKey = $orderCode !== '' ? $orderCode : 'FP' . md5(implode('|', [
|
|
$c['quota_id'] ?? '', $c['pay_time'] ?? '', $c['consume_amount'] ?? '',
|
|
$c['pay_fee'] ?? '', $c['station_code'] ?? '', $c['litter'] ?? '',
|
|
]));
|
|
$consumeId = 0;
|
|
try {
|
|
// 流水落库与订单累计放在同一事务：唯一键挡住重推，重推既不重复累计也不重复通知
|
|
Db::transaction(function () use ($order, $c, $code, $orderCode, $consumeKey, $payTime, $consumeFen, $consumeYuan, &$consumeId) {
|
|
$consumeId = (int)Db::name('qyt_consume')->insertGetId([
|
|
'order_id' => $order['order_id'],
|
|
'waybill_no' => $c['waybill_no'] ?? '',
|
|
'waybill_no_code' => $code,
|
|
'order_code' => $orderCode,
|
|
'consume_key' => $consumeKey,
|
|
'station_code' => $c['station_code'] ?? '',
|
|
'station_name' => $c['station_name'] ?? '',
|
|
'oil_name' => $c['oil_name'] ?? '',
|
|
'litter' => $c['litter'] ?? '',
|
|
'oil_price' => (int)($c['oil_price'] ?? 0),
|
|
'actual_oil_price' => (int)($c['actual_oil_price'] ?? 0),
|
|
'pay_fee' => (int)($c['pay_fee'] ?? 0),
|
|
'consume_amount' => $consumeFen,
|
|
'pay_time' => $payTime,
|
|
'raw' => json_encode($c, 320),
|
|
'store_id' => $order['store_id'],
|
|
'create_time' => time(),
|
|
]);
|
|
// 累计已加油金额 / 次数，并记下最近一次加油信息（原子自增，不读后写）
|
|
Db::name('partners_order')->where('order_id', $order['order_id'])->update([
|
|
'used_amount' => Db::raw('`used_amount` + ' . $consumeYuan),
|
|
'used_num' => Db::raw('`used_num` + 1'),
|
|
'use_text' => '加油',
|
|
'qyt_order_code' => $orderCode,
|
|
'station_name' => $c['station_name'] ?? '',
|
|
'oil_name' => $c['oil_name'] ?? '',
|
|
'litter' => $c['litter'] ?? '',
|
|
'pay_fee' => (int)($c['pay_fee'] ?? 0),
|
|
'consume_time' => $payTime,
|
|
'update_time' => time(),
|
|
]);
|
|
// use_status 语义不变（0 未加油 / 1 有加油记录）：只在第一笔加油时置 1 并记首次加油时间
|
|
Db::name('partners_order')->where('order_id', $order['order_id'])->where('use_status', 0)->update([
|
|
'use_status' => 1,
|
|
'use_time' => time(),
|
|
'shareNo' => $orderCode !== '' ? $orderCode : $order['shareNo'],
|
|
]);
|
|
});
|
|
} catch (\Exception $e) {
|
|
if (strpos($e->getMessage(), 'Duplicate') === false) {
|
|
throw $e;
|
|
}
|
|
$SybOil->saveLogStore('企油通消费回调重复，忽略（不重复累计、不重复通知）：' . $code . '/' . $consumeKey);
|
|
continue;
|
|
}
|
|
|
|
// 超额告警：累计加油 + 已退款 > 订单金额。上游额度按司机凭证累计，
|
|
// 归属到本单的消费可能与本地可退金额算出的不一致，需要人工核对
|
|
$fresh = Db::name('partners_order')->where('order_id', $order['order_id'])
|
|
->field('total_price,used_amount,refund_amount,oil_status_zsy')->find();
|
|
if ($fresh && (float)$fresh['used_amount'] + (float)$fresh['refund_amount'] > (float)$fresh['total_price'] + 0.005) {
|
|
$SybOil->saveLogStore('【告警】订单已加油+已退款超出订单金额 order=' . $order['order_id']
|
|
. ' total=' . $fresh['total_price'] . ' used=' . $fresh['used_amount'] . ' refund=' . $fresh['refund_amount']);
|
|
}
|
|
|
|
// 每笔加油都通知下单平台（amount = 本次占用金额）
|
|
$notify = json_encode([
|
|
'order_id' => $order['order_id'],
|
|
'optType' => 'add_oil',
|
|
'consume_id' => $consumeId,
|
|
'retry_count' => 0,
|
|
], 320);
|
|
$queueService = new QueueService();
|
|
if ((int)($fresh['oil_status_zsy'] ?? 2) === 2) {
|
|
$queueService->push('notify_queue', $notify);
|
|
} else {
|
|
// 7.8 发放结果还没确认就先收到了加油：延后 60 秒再通知，尽量让「分油成功」通知先到下游
|
|
$queueService->laterPush('notify_queue', $notify, 60);
|
|
}
|
|
$SybOil->saveLogStore('企油通消费回调处理成功：' . json_encode($c, 320));
|
|
}
|
|
break;
|
|
default:
|
|
$SybOil->saveLogStore('队列数据处理异常，未知：【'.$message.'】');
|
|
}
|
|
} catch (\Exception $e) {
|
|
$SybOil->saveLogStore(
|
|
'队列回调处理异常：' . $e->getMessage() .
|
|
' 在文件：' . $e->getFile() .
|
|
' 第：' . $e->getLine() .
|
|
' 行。堆栈跟踪：' . $e->getTraceAsString()
|
|
);
|
|
|
|
}
|
|
}
|
|
}
|