add ids args

backend/app/Command/AppMessageQueuePushTmall.php

@@ -35,6 +35,12 @@ class AppMessageQueuePushTmall extends HyperfCommand
             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'
+        );
     }
 
     /**
@@ -47,16 +53,19 @@ class AppMessageQueuePushTmall extends HyperfCommand
                 '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'
             ]
         ];
 
@@ -78,10 +87,19 @@ class AppMessageQueuePushTmall extends HyperfCommand
             $this->info(sprintf('Processing queue type: %s, table: %s', $queueType, $config['table']));
 
             // 从 raw 数据库连接获取数据
-            $records = Db::connection('raw')
-                ->table($config['table'])
-                ->orderBy('id', 'desc')
-                ->limit(4)->get($config['column'])->lazy();
+            $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']));