datahub/backend/app/Command/AppMessageQueuePushTmall.php

<?php

declare(strict_types=1);

namespace App\Command;

use Hyperf\Collection\LazyCollection;
use Hyperf\Command\Command as HyperfCommand;
use Hyperf\Command\Annotation\Command;
use Psr\Container\ContainerInterface;
use Hyperf\DbConnection\Db;
use App\Platform\Tmall\Producer\TmallOrderProducer;
use App\Platform\Tmall\Producer\TmallProductProducer;
use App\Platform\Tmall\Producer\TmallRefundProducer;
use App\Platform\Tmall\Producer\TmallInventoryProducer;
use Hyperf\Amqp\Producer;
use Symfony\Component\Console\Input\InputOption;
use Exception;

#[Command]
class AppMessageQueuePushTmall extends HyperfCommand
{
    public function __construct(protected ContainerInterface $container)
    {
        parent::__construct('app:mq-push:tmall');
    }

    public function configure()
    {
        parent::configure();
        $this->setDescription('Test push message with Tmall km data');
        $this->addOption(
            'queue-type',
            't',
            InputOption::VALUE_REQUIRED,
            'Queue type: product, order, refund, inventory'
        );
        $this->addOption(
            'ids',
            null,
            InputOption::VALUE_REQUIRED,
            'Comma-separated primary key IDs to fetch specific records'
        );
    }

    /**
     * 获取队列类型配置
     */
    protected function getQueueTypeConfig(string $type): array
    {
        $configs = [
            'order' => [
                'table' => 'wpic_taobao_order',
                'column' => 'order_raw',
                'producer' => TmallOrderProducer::class,
                'primary' => 'tid'
            ],
            'product' => [
                'table' => 'wpic_taobao_item',
                'column' => 'item_raw',
                'producer' => TmallProductProducer::class,
                 'primary' => 'iid'
            ],
            'refund' => [
                'table' => 'wpic_taobao_return_item',
                'column' => 'refund_raw',
                'producer' => TmallRefundProducer::class,
                'primary' => 'refund_id'
            ]
        ];

        return $configs[$type] ?? throw new Exception("Invalid queue type: {$type}");
    }

    public function handle(): void
    {
        try {
            $queueType = $this->input->getOption('queue-type');
            $validTypes = ['product', 'order', 'refund', 'inventory'];

            if (empty($queueType) || !in_array($queueType, $validTypes)) {
                $this->error(sprintf('--queue-type is required. Must be one of: %s', implode(', ', $validTypes)));
                return;
            }

            $config = $this->getQueueTypeConfig($queueType);
            $this->info(sprintf('Processing queue type: %s, table: %s', $queueType, $config['table']));

            // 从 raw 数据库连接获取数据
            $query = Db::connection('raw')->table($config['table']);

            $idsOption = $this->input->getOption('ids');
            if (!empty($idsOption)) {
                $ids = array_map('trim', explode(',', $idsOption));
                $ids = array_unique(array_filter($ids, fn($id) => $id !== ''));
                $query->whereIn('id', $ids);
                $this->info(sprintf('Fetching records by %s: %s', $config['primary'], implode(', ', $ids)));
            } else {
                $query->orderBy('id', 'desc')->limit(4);
            }

            $records = $query->get([$config['column']])->lazy();

            if ($records->isEmpty()) {
                $this->warn(sprintf('No records found in %s table', $config['table']));
                return;
            }

            $this->info(sprintf('Found %d records, processing...', $records->count()));

            // 获取 Producer 实例
            $producer = $this->container->get(Producer::class);
            $producerClass = $config['producer'];

            $messageCount = 0;

            $records->chunk(2)->each(function (LazyCollection $collection) use ($producer, $producerClass, $config, &$messageCount) {

                $raw_data = $collection->pluck($config['column'])->map(function ($item) {
                    return json_decode($item, true);
                })->toArray();

                //@ATTENTION 生产环境需要注意， 暂时使用 KM 进行测试
                $messageData = [
                    'company_id' => 188,
                    'platform_id' => 2,
                    'store_id' => 292,
                    'unique_id' =>  time() . '_' . uniqid(),
                    'raw_data' => $raw_data, // 包含 2 条原始记录
                ];

                // 创建并发送消息
                $message = new $producerClass($messageData);
                $producer->produce($message);

                $messageCount++;

                $this->line(sprintf('Sent message %d with unique ID: %s',
                    $messageCount,
                    $messageData['unique_id'],
                ));
            });


            $this->info(sprintf('Successfully sent %d messages to RabbitMQ', $messageCount));
            return;

        } catch (Exception $e) {
            $this->error('Error pushing messages: ' . $e->getMessage());
            $this->error($e->getTraceAsString());
            return;
        }
    }
}