245 lines
12 KiB
PHP
245 lines
12 KiB
PHP
<?php
|
||
// +----------------------------------------------------------------------
|
||
// | 下单平台订单 -> 企油通「运单油费下发」(7.7)
|
||
// +----------------------------------------------------------------------
|
||
// | OilCard::share 创建 yoshop_partners_order 后 push 'partners_queue'，本命令消费：
|
||
// | 1. 每个 order_id 用 RedisLock 串行化，锁内重读订单状态
|
||
// | 2. 生成企油通全平台唯一编号（确定性派生，可安全重试）
|
||
// | 3. 调 7.7 createAndSendQuota
|
||
// | 4. 受理成功 -> oil_status_zsy=1「发放中」，交给 qyt:status 轮询 7.8 定结果
|
||
// |
|
||
// | ⚠️ 7.7 是异步接口，返回成功只代表【受理】（文档 §5 第 7 条）。本命令不再直接
|
||
// | 判定发放成功，也不再投 notify_queue —— 那是 QytStatusQueue 拿到 7.8 结果后的事。
|
||
// |
|
||
// | 并发/原子性：
|
||
// | · oil_status_zsy 只在 7.7 结果确定后离开 0；空闲时 reconcile() 补投丢失的队列消息
|
||
// | · 状态跃迁为条件 UPDATE(... WHERE oil_status_zsy=0)，affected=1 才继续，天然幂等
|
||
// | · batch_no 确定性派生 -> 崩溃重试用同一值，依赖企油通按 batch_no 幂等去重
|
||
// | 启动：php think partners:queue（可多进程；RedisLock 保证同一订单串行）
|
||
// +----------------------------------------------------------------------
|
||
namespace app\console\command;
|
||
|
||
use app\common\lock\RedisLock;
|
||
use app\common\model\partners\Order as OrderModel;
|
||
use app\common\model\qyt\Enterprise as EnterpriseModel;
|
||
use app\common\service\PartnersBalance;
|
||
use app\common\service\QueueService;
|
||
use app\common\service\QytOil;
|
||
use think\console\Command;
|
||
use think\console\Input;
|
||
use think\console\Output;
|
||
use think\facade\Db;
|
||
|
||
class PartnersQueue extends Command
|
||
{
|
||
/** @var QueueService */
|
||
protected $queueService;
|
||
|
||
/** @var int 上次兜底巡检时间 */
|
||
protected $lastReconcile = 0;
|
||
|
||
protected function configure()
|
||
{
|
||
$this->setName('partners:queue')
|
||
->setDescription('下单平台订单 -> 企油通运单油费下发(7.7)');
|
||
}
|
||
|
||
protected function execute(Input $input, Output $output)
|
||
{
|
||
$this->queueService = new QueueService();
|
||
while (true) {
|
||
$message = $this->queueService->pop('partners_queue');
|
||
if ($message) {
|
||
$this->processMessage($message);
|
||
} else {
|
||
$this->reconcile();
|
||
sleep(1);
|
||
}
|
||
}
|
||
}
|
||
|
||
protected function processMessage($message)
|
||
{
|
||
$qyt = new QytOil();
|
||
$data = is_array($message) ? $message : json_decode($message, true);
|
||
$orderId = $data['order_id'] ?? null;
|
||
if (!$orderId) {
|
||
$qyt->saveLog('partners_queue 消息缺少 order_id：' . json_encode($message, 320));
|
||
return;
|
||
}
|
||
|
||
// 每个订单串行处理；锁 60s 自动过期，崩溃不会死锁
|
||
$lock = new RedisLock('qyt:send:' . $orderId, 60);
|
||
if (!$lock->acquireLock()) {
|
||
$qyt->saveLog("partners_queue 订单处理中，跳过 order={$orderId}");
|
||
return;
|
||
}
|
||
|
||
try {
|
||
$order = OrderModel::detail($orderId);
|
||
if (!$order) {
|
||
$qyt->saveLog('partners_queue 未找到订单：' . $orderId);
|
||
return;
|
||
}
|
||
if ($order['order_status'] != 10 || $order['oil_status'] != 0 || (int) $order['oil_status_zsy'] !== 0) {
|
||
$qyt->saveLog("partners_queue 订单状态不满足下发条件 order={$orderId}");
|
||
return;
|
||
}
|
||
|
||
// 用下单时固化在订单上的企业，而不是当前默认企业：
|
||
// 之后改了合作商绑定或切了默认企业，在途订单仍按原企业发放
|
||
$enterprise = EnterpriseModel::detail((int) $order['qyt_enterprise_id']);
|
||
if (!$enterprise) {
|
||
$failed = Db::name('partners_order')->where('order_id', $orderId)->where('oil_status_zsy', 0)
|
||
->update(['oil_status_zsy' => 3, 'oil_text' => '本单绑定的企油通企业不存在，无法发放', 'update_time' => time()]);
|
||
$failed && $this->refundBalance($order);
|
||
return;
|
||
}
|
||
$qyt->useEnterprise($enterprise);
|
||
|
||
// 1. 编号：库里已生成过就直接沿用，绝不重算。
|
||
// no_prefix 现在可在后台修改，若重试时重新派生，batch_no 会变成另一个值，
|
||
// 上游会当成新批次 -> 同一订单被发放两次。
|
||
// 种子里带上重发次数：后台「重新发放」会清空编号并把 qyt_retry_num +1，
|
||
// 这样重发能拿到全新的 batch_no（否则企油通按老批次幂等，7.8 永远返回那条旧的失败记录）。
|
||
$retryNum = (int) $order['qyt_retry_num'];
|
||
$seed = ($order['partners_order_no'] ?: (string) $order['order_id'])
|
||
. ($retryNum > 0 ? 'R' . $retryNum : '');
|
||
$batchNo = $order['batch_no'] ?: $qyt->no('B', $seed);
|
||
$waybillNo = $order['waybill_no'] ?: $qyt->no('W', $seed);
|
||
$waybillNoCode = $order['waybill_no_code'] ?: $qyt->no('C', $seed);
|
||
$name = trim((string) $order['driver_name']);
|
||
$idCard = trim((string) $order['id_card']);
|
||
// 两个通道都要求实名，姓名/手机号/身份证号缺一不可。
|
||
// 这类单直接判失败退款，不做无意义重试（重试也永远补不齐这些字段），
|
||
// 典型来源是改造前建的历史单，它们没有 id_card。
|
||
$missing = [];
|
||
$name === '' && $missing[] = '司机姓名';
|
||
$order['mobile'] === '' && $missing[] = '手机号';
|
||
$idCard === '' && $missing[] = '身份证号';
|
||
if ($missing) {
|
||
$failed = Db::name('partners_order')->where('order_id', $orderId)->where('oil_status_zsy', 0)
|
||
->update([
|
||
'oil_status_zsy' => 3,
|
||
'oil_text' => '缺' . implode('/', $missing) . '，无法发放',
|
||
'update_time' => time(),
|
||
]);
|
||
$failed && $this->refundBalance($order);
|
||
return;
|
||
}
|
||
|
||
// 2. 先落编号 + 本单实际使用的支付通道（状态不动，仍为 0）
|
||
// 通道随单落库：切换通道后老单仍能追溯当时走的是哪条通道
|
||
Db::name('partners_order')->where('order_id', $orderId)->update([
|
||
'batch_no' => $batchNo,
|
||
'waybill_no' => $waybillNo,
|
||
'waybill_no_code' => $waybillNoCode,
|
||
'passageway' => $qyt->channelKey(),
|
||
'oil_text' => '发放中(' . $qyt->channelName() . ')',
|
||
'update_time' => time(),
|
||
]);
|
||
|
||
// 3. 7.7 运单油费下发
|
||
$res = $qyt->createAndSendQuota([
|
||
'batch_no' => $batchNo,
|
||
'waybill_no' => $waybillNo,
|
||
'waybill_no_code' => $waybillNoCode,
|
||
'mobile' => $order['mobile'],
|
||
'name' => $name,
|
||
'id_card' => $idCard,
|
||
'amount' => $order['total_price'],
|
||
]);
|
||
|
||
// 失败（且非 batch 已存在）
|
||
if ($res['code'] !== '00' && empty($res['dup_batch'])) {
|
||
$failed = Db::name('partners_order')->where('order_id', $orderId)->where('oil_status_zsy', 0)->update([
|
||
'oil_status_zsy' => 3,
|
||
'oil_text' => '下发失败：' . $res['message'],
|
||
'update_time' => time(),
|
||
]);
|
||
$failed && $this->refundBalance($order);
|
||
$qyt->saveLog("7.7 下发失败 order={$orderId}：" . $res['message']);
|
||
return;
|
||
}
|
||
// data.status: 1 成功 / 2 失败（dup_batch 视为已受理成功）
|
||
$status = empty($res['dup_batch']) ? (string) ($res['data']['status'] ?? '1') : '1';
|
||
if ($status === '2') {
|
||
$failed = Db::name('partners_order')->where('order_id', $orderId)->where('oil_status_zsy', 0)->update([
|
||
'oil_status_zsy' => 3,
|
||
'oil_text' => $res['data']['msg'] ?? '下发失败',
|
||
'update_time' => time(),
|
||
]);
|
||
$failed && $this->refundBalance($order);
|
||
return;
|
||
}
|
||
|
||
// 4. 受理成功【不等于】发放成功：7.7 是异步接口，真实结果要用 7.8 按 batch_no 查。
|
||
// 这里只把状态推进到「发放中」，由 qyt:status 轮询 7.8 后才决定成功/失败并回调下游。
|
||
$affected = Db::name('partners_order')->where('order_id', $orderId)->where('oil_status_zsy', 0)->update([
|
||
'oil_status_zsy' => 1,
|
||
'shareNo' => $waybillNoCode,
|
||
'oil_text' => '已受理，等待企油通发放结果'
|
||
. (empty($res['dup_batch']) ? '' : '（batch 已存在，按已受理处理）'),
|
||
'qyt_query_time' => 0,
|
||
'qyt_query_num' => 0,
|
||
'update_time' => time(),
|
||
]);
|
||
if ($affected) {
|
||
$qyt->saveLog("7.7 已受理 order={$orderId} batch_no={$batchNo} 通道={$qyt->channelName()}"
|
||
. (empty($res['dup_batch']) ? '' : '（batch 已存在，按已受理处理）'));
|
||
}
|
||
} catch (\Throwable $e) {
|
||
// 不改状态（仍为 0），交由 reconcile 兜底重投
|
||
$qyt->saveLog('partners_queue 异常 order=' . $orderId . '：' . $e->getMessage()
|
||
. ' @ ' . $e->getFile() . ':' . $e->getLine());
|
||
} finally {
|
||
$lock->releaseLock();
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 发放失败 -> 回补合作商余额（仅在状态跃迁胜出时调用，天然幂等）
|
||
*/
|
||
protected function refundBalance($order)
|
||
{
|
||
PartnersBalance::refund(
|
||
(int) $order['partners_id'],
|
||
(float) $order['total_price'],
|
||
(string) $order['partners_order_no'],
|
||
(int) $order['order_id'],
|
||
(int) $order['store_id']
|
||
);
|
||
(new QytOil())->saveLog("发放失败已回补合作商余额 order={$order['order_id']} amount={$order['total_price']}");
|
||
}
|
||
|
||
/**
|
||
* 兜底巡检（限频 60s）：补投丢失的 partners_queue 消息。
|
||
* 因下发为条件跃迁，重复投递安全。
|
||
*/
|
||
protected function reconcile()
|
||
{
|
||
if (time() - $this->lastReconcile < 60) {
|
||
return;
|
||
}
|
||
$this->lastReconcile = time();
|
||
$now = time();
|
||
try {
|
||
$ids = Db::name('partners_order')
|
||
->where('order_status', 10)
|
||
->where('oil_status_zsy', 0)
|
||
->where('oilCardType', 'zsy_ticket')
|
||
->where('create_time', '<', $now - 120)
|
||
->where('create_time', '>', $now - 86400)
|
||
->limit(200)->column('order_id');
|
||
foreach ($ids as $oid) {
|
||
$this->queueService->push('partners_queue', json_encode(['order_id' => $oid], 320));
|
||
}
|
||
if ($ids) {
|
||
(new QytOil())->saveLog('partners_queue reconcile 补投 ' . count($ids) . ' 单');
|
||
}
|
||
} catch (\Throwable $e) {
|
||
(new QytOil())->saveLog('partners_queue reconcile 异常：' . $e->getMessage());
|
||
}
|
||
}
|
||
}
|