fix: change backfill product image

api_v1/management/commands/backfill_external_product_images.py

@@ -6,12 +6,12 @@ from basic_info.models import Product
 
 
 class Command(BaseCommand):
-    help = '为没有图片的 Product 通过外部 image API 回补图片（使用 name_b64=base64url(product.name)）'
+    help = '为 image 为空但 get_primary_image_url() 非空的 Product 通过外部 image API 回补图片'
 
     def add_arguments(self, parser):
         parser.add_argument('--merchant-id', type=int, help='仅处理指定商户的产品')
         parser.add_argument('--product-id', type=int, help='仅处理指定产品')
-        parser.add_argument('--limit', type=int, help='最多处理多少条产品')
+        parser.add_argument('--limit', type=int, help='最多处理多少条符合条件的产品')
         parser.add_argument(
             '--dry-run',
             action='store_true',
@@ -41,50 +41,65 @@ class Command(BaseCommand):
         if limit is not None:
             if int(limit) <= 0:
                 raise CommandError('--limit 必须大于 0')
-            queryset = queryset[: int(limit)]
+            limit = int(limit)
-
-        products = list(queryset)
-        self.stdout.write(f'待处理产品数: {len(products)}')
-
-        if options.get('dry_run'):
-            for product in products:
-                self.stdout.write(
-                    f'[DRY-RUN] product_id={product.id} merchant_id={product.merchant_id} name={product.name}'
-                )
-            self.stdout.write(self.style.SUCCESS('dry-run 完成'))
-            return
 
         success_count = 0
         failed_count = 0
         skipped_count = 0
+        skipped_no_primary_count = 0
+        selected_count = 0
+        scanned_count = 0
+        dry_run = bool(options.get('dry_run'))
 
-        for product in products:
+        for product in queryset.iterator(chunk_size=1000):
+            scanned_count += 1
             if product.image:
                 skipped_count += 1
                 continue
 
-            try:
+            if not product.get_primary_image_url():
-                image_payload = _fetch_external_product_image(product.name)
+                skipped_no_primary_count += 1
-                _upload_product_image(product, image_payload)
-            except Exception as exc:
-                failed_count += 1
-                self.stderr.write(
-                    self.style.ERROR(
-                        f'补图失败: product_id={product.id} merchant_id={product.merchant_id} '
-                        f'name={product.name} error={exc}'
-                    )
-                )
                 continue
 
-            success_count += 1
+            selected_count += 1
-            self.stdout.write(
+            if dry_run:
-                self.style.SUCCESS(
+                self.stdout.write(
-                    f'补图成功: product_id={product.id} merchant_id={product.merchant_id} name={product.name}'
+                    f'[DRY-RUN] product_id={product.id} merchant_id={product.merchant_id} name={product.name}'
                 )
-            )
+            else:
+                try:
+                    image_payload = _fetch_external_product_image(product.name)
+                    _upload_product_image(product, image_payload)
+                except Exception as exc:
+                    failed_count += 1
+                    self.stderr.write(
+                        self.style.ERROR(
+                            f'补图失败: product_id={product.id} merchant_id={product.merchant_id} '
+                            f'name={product.name} error={exc}'
+                        )
+                    )
+                else:
+                    success_count += 1
+                    self.stdout.write(
+                        self.style.SUCCESS(
+                            f'补图成功: product_id={product.id} merchant_id={product.merchant_id} name={product.name}'
+                        )
+                    )
+
+            if limit is not None and selected_count >= limit:
+                break
+
+        if dry_run:
+            self.stdout.write(self.style.SUCCESS('dry-run 完成'))
 
         self.stdout.write(
             self.style.SUCCESS(
-                f'完成: total={len(products)} success={success_count} failed={failed_count} skipped={skipped_count}'
+                '完成: '
+                f'scanned={scanned_count} '
+                f'selected={selected_count} '
+                f'success={success_count} '
+                f'failed={failed_count} '
+                f'skipped_with_image={skipped_count} '
+                f'skipped_without_primary_url={skipped_no_primary_count}'
             )
         )

api_v1/test_backfill_external_product_images_command.py

@@ -0,0 +1,166 @@
+import json
+from io import StringIO
+from unittest.mock import patch
+
+from django.core.management import call_command
+from django.test import TestCase
+
+from basic_info import models as basic_models
+
+
+class BackfillExternalProductImagesCommandTest(TestCase):
+    def setUp(self):
+        self.merchant = basic_models.Merchant.objects.create(
+            name='测试商户',
+            type=basic_models.MerchantTypeEnum.FACTORY,
+        )
+        self.other_merchant = basic_models.Merchant.objects.create(
+            name='其他商户',
+            type=basic_models.MerchantTypeEnum.FACTORY,
+        )
+        self.category = basic_models.ProductCategory.objects.create(
+            merchant=self.merchant,
+            name='测试分类',
+            product_prefix='TP',
+        )
+        self.other_category = basic_models.ProductCategory.objects.create(
+            merchant=self.other_merchant,
+            name='其他分类',
+            product_prefix='OP',
+        )
+
+    def _create_product(self, *, merchant, category, name, image=None, description=None):
+        return basic_models.Product.objects.create(
+            merchant=merchant,
+            category=category,
+            name=name,
+            image=image,
+            description=description,
+        )
+
+    @patch('api_v1.management.commands.backfill_external_product_images._upload_product_image')
+    @patch('api_v1.management.commands.backfill_external_product_images._fetch_external_product_image')
+    def test_command_only_processes_products_with_primary_image_url(
+        self,
+        mock_fetch_image,
+        mock_upload_image,
+    ):
+        eligible = self._create_product(
+            merchant=self.merchant,
+            category=self.category,
+            name='有图源产品#1',
+            description=json.dumps({'original_file_full_path': 'https://example.com/a.jpg'}),
+        )
+        self._create_product(
+            merchant=self.merchant,
+            category=self.category,
+            name='无图源产品#2',
+            description=None,
+        )
+        self._create_product(
+            merchant=self.merchant,
+            category=self.category,
+            name='已有图片产品#3',
+            image='product_images/existing.jpg',
+            description=json.dumps({'original_file_full_path': 'https://example.com/existing.jpg'}),
+        )
+        self._create_product(
+            merchant=self.other_merchant,
+            category=self.other_category,
+            name='其他商户产品#4',
+            description=json.dumps({'original_file_full_path': 'https://example.com/other.jpg'}),
+        )
+
+        mock_fetch_image.return_value = {
+            'bytes': b'fake-image',
+            'content_type': 'image/jpeg',
+            'name': 'from-api.jpg',
+        }
+
+        stdout = StringIO()
+        call_command(
+            'backfill_external_product_images',
+            '--merchant-id',
+            str(self.merchant.id),
+            stdout=stdout,
+        )
+
+        mock_fetch_image.assert_called_once_with(eligible.name)
+        self.assertEqual(mock_upload_image.call_count, 1)
+        self.assertIn('selected=1', stdout.getvalue())
+        self.assertIn('skipped_without_primary_url=1', stdout.getvalue())
+
+    @patch('api_v1.management.commands.backfill_external_product_images._upload_product_image')
+    @patch('api_v1.management.commands.backfill_external_product_images._fetch_external_product_image')
+    def test_limit_applies_after_primary_url_filtering(
+        self,
+        mock_fetch_image,
+        mock_upload_image,
+    ):
+        self._create_product(
+            merchant=self.merchant,
+            category=self.category,
+            name='无图源产品#0',
+            description=None,
+        )
+        self._create_product(
+            merchant=self.merchant,
+            category=self.category,
+            name='有图源产品#1',
+            description=json.dumps({'original_file_full_path': 'https://example.com/1.jpg'}),
+        )
+        eligible_two = self._create_product(
+            merchant=self.merchant,
+            category=self.category,
+            name='有图源产品#2',
+            description=json.dumps({'original_file_full_path': 'https://example.com/2.jpg'}),
+        )
+
+        mock_fetch_image.return_value = {
+            'bytes': b'fake-image',
+            'content_type': 'image/jpeg',
+            'name': 'from-api.jpg',
+        }
+
+        stdout = StringIO()
+        call_command(
+            'backfill_external_product_images',
+            '--merchant-id',
+            str(self.merchant.id),
+            '--limit',
+            '1',
+            stdout=stdout,
+        )
+
+        mock_fetch_image.assert_called_once_with(eligible_two.name)
+        self.assertEqual(mock_upload_image.call_count, 1)
+        self.assertIn('selected=1', stdout.getvalue())
+        self.assertIn('success=1', stdout.getvalue())
+
+    def test_dry_run_lists_only_eligible_products(self):
+        eligible = self._create_product(
+            merchant=self.merchant,
+            category=self.category,
+            name='有图源产品#dry',
+            description=json.dumps({'original_file_full_path': 'https://example.com/dry.jpg'}),
+        )
+        self._create_product(
+            merchant=self.merchant,
+            category=self.category,
+            name='无图源产品#dry',
+            description=None,
+        )
+
+        stdout = StringIO()
+        call_command(
+            'backfill_external_product_images',
+            '--merchant-id',
+            str(self.merchant.id),
+            '--dry-run',
+            stdout=stdout,
+        )
+
+        output = stdout.getvalue()
+        self.assertIn(eligible.name, output)
+        self.assertNotIn('无图源产品#dry', output)
+        self.assertIn('dry-run 完成', output)