1
0
forked from erp-dev/erp

feat: backfill external order image upgrade to beat schedule by 1 hour

This commit is contained in:
2026-03-31 18:10:24 +08:00
parent aff10b8925
commit 72840b7277
12 changed files with 1085 additions and 108 deletions

View File

@@ -0,0 +1,136 @@
from collections.abc import Callable
from django.db.models import Q
from printing.models import PrintingJob
def extract_request_name(job: PrintingJob) -> str:
raw = job.external_raw or {}
if isinstance(raw, dict):
request_name = str(raw.get("YanSe") or "").strip()
if request_name:
return request_name
return str(job.external_product_name or "").strip()
def run_external_product_image_backfill(
*,
fetch_image: Callable[[str], dict],
upload_image: Callable[[object, dict], None],
merchant_id: int | None = None,
product_id: int | None = None,
limit: int | None = None,
dry_run: bool = False,
on_success: Callable[[str], None] | None = None,
on_error: Callable[[str], None] | None = None,
on_info: Callable[[str], None] | None = None,
) -> dict:
queryset = (
PrintingJob.objects.filter(
printing_order__external_order_id__isnull=False,
)
.exclude(printing_order__external_order_id="")
.filter(Q(product__image__isnull=True) | Q(product__image=""))
.select_related("product", "printing_order", "product__merchant")
.order_by("-id")
)
if merchant_id:
queryset = queryset.filter(product__merchant_id=merchant_id)
if product_id:
queryset = queryset.filter(product_id=product_id)
if limit is not None:
limit = int(limit)
if limit <= 0:
raise ValueError("limit must be greater than 0")
success_count = 0
failed_count = 0
skipped_with_image_count = 0
skipped_without_request_name_count = 0
skipped_duplicate_product_count = 0
selected_count = 0
scanned_count = 0
seen_product_ids = set()
for job in queryset.iterator(chunk_size=1000):
scanned_count += 1
product = job.product
if product.image:
skipped_with_image_count += 1
continue
if product.id in seen_product_ids:
skipped_duplicate_product_count += 1
continue
request_name = extract_request_name(job)
if not request_name:
skipped_without_request_name_count += 1
continue
seen_product_ids.add(product.id)
selected_count += 1
if dry_run:
if on_info:
on_info(
"[DRY-RUN] "
f"job_id={job.id} "
f"product_id={product.id} "
f"merchant_id={product.merchant_id} "
f"product_name={product.name} "
f"request_name={request_name}"
)
else:
try:
image_payload = fetch_image(request_name)
upload_image(product, image_payload)
except Exception as exc:
failed_count += 1
if on_error:
on_error(
"补图失败: "
f"job_id={job.id} "
f"product_id={product.id} "
f"merchant_id={product.merchant_id} "
f"product_name={product.name} "
f"request_name={request_name} "
f"error={exc}"
)
else:
success_count += 1
if on_success:
on_success(
"补图成功: "
f"job_id={job.id} "
f"product_id={product.id} "
f"merchant_id={product.merchant_id} "
f"product_name={product.name} "
f"request_name={request_name}"
)
if limit is not None and selected_count >= limit:
break
payload = {
"scanned": scanned_count,
"selected": selected_count,
"success": success_count,
"failed": failed_count,
"skipped_with_image": skipped_with_image_count,
"skipped_without_request_name": skipped_without_request_name_count,
"skipped_duplicate_product": skipped_duplicate_product_count,
"merchant_id": merchant_id,
"product_id": product_id,
"limit": limit,
"dry_run": dry_run,
}
if dry_run and on_info:
on_info("dry-run 完成")
return payload

View File

@@ -1,8 +1,7 @@
from django.core.management.base import BaseCommand, CommandError
from django.db.models import Q
from api_v1.external_product_image_backfill import run_external_product_image_backfill
from api_v1.tasks import _fetch_external_product_image, _upload_product_image
from printing.models import PrintingJob
class Command(BaseCommand):
@@ -18,114 +17,25 @@ class Command(BaseCommand):
help='仅输出将处理的产品,不实际请求外部 API 或写库',
)
@staticmethod
def _extract_request_name(job: PrintingJob) -> str:
raw = job.external_raw or {}
if isinstance(raw, dict):
request_name = str(raw.get('YanSe') or '').strip()
if request_name:
return request_name
return str(job.external_product_name or '').strip()
def handle(self, *args, **options):
queryset = (
PrintingJob.objects.filter(
printing_order__external_order_id__isnull=False,
)
.exclude(printing_order__external_order_id='')
.filter(Q(product__image__isnull=True) | Q(product__image=''))
.select_related('product', 'printing_order', 'product__merchant')
.order_by('-id')
)
merchant_id = options.get('merchant_id')
if merchant_id:
queryset = queryset.filter(product__merchant_id=merchant_id)
product_id = options.get('product_id')
if product_id:
queryset = queryset.filter(product_id=product_id)
limit = options.get('limit')
if limit is not None:
if int(limit) <= 0:
raise CommandError('--limit 必须大于 0')
limit = int(limit)
success_count = 0
failed_count = 0
skipped_with_image_count = 0
skipped_without_request_name_count = 0
skipped_duplicate_product_count = 0
selected_count = 0
scanned_count = 0
dry_run = bool(options.get('dry_run'))
seen_product_ids = set()
for job in queryset.iterator(chunk_size=1000):
scanned_count += 1
product = job.product
if product.image:
skipped_with_image_count += 1
continue
if product.id in seen_product_ids:
skipped_duplicate_product_count += 1
continue
request_name = self._extract_request_name(job)
if not request_name:
skipped_without_request_name_count += 1
continue
seen_product_ids.add(product.id)
selected_count += 1
if dry_run:
self.stdout.write(
'[DRY-RUN] '
f'job_id={job.id} '
f'product_id={product.id} '
f'merchant_id={product.merchant_id} '
f'product_name={product.name} '
f'request_name={request_name}'
)
else:
try:
image_payload = _fetch_external_product_image(request_name)
_upload_product_image(product, image_payload)
except Exception as exc:
failed_count += 1
self.stderr.write(
self.style.ERROR(
f'补图失败: job_id={job.id} product_id={product.id} merchant_id={product.merchant_id} '
f'product_name={product.name} request_name={request_name} error={exc}'
)
)
else:
success_count += 1
self.stdout.write(
self.style.SUCCESS(
f'补图成功: job_id={job.id} product_id={product.id} merchant_id={product.merchant_id} '
f'product_name={product.name} request_name={request_name}'
)
payload = run_external_product_image_backfill(
fetch_image=_fetch_external_product_image,
upload_image=_upload_product_image,
merchant_id=merchant_id,
product_id=product_id,
limit=limit,
dry_run=bool(options.get('dry_run')),
on_success=lambda message: self.stdout.write(self.style.SUCCESS(message)),
on_error=lambda message: self.stderr.write(self.style.ERROR(message)),
on_info=lambda message: self.stdout.write(message),
)
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'scanned={scanned_count} '
f'selected={selected_count} '
f'success={success_count} '
f'failed={failed_count} '
f'skipped_with_image={skipped_with_image_count} '
f'skipped_without_request_name={skipped_without_request_name_count} '
f'skipped_duplicate_product={skipped_duplicate_product_count}'
)
)
self.stdout.write(self.style.SUCCESS(f'完成: {payload}'))

View File

@@ -31,6 +31,7 @@ from api_v1.mdy_plate_order_staging_tiia_upload import (
build_default_tiia_rate_limiter,
upload_mdy_plate_order_staging_plate_images_to_tencent_tiia,
)
from api_v1.external_product_image_backfill import run_external_product_image_backfill
from printing import models as printing_models
@@ -424,6 +425,37 @@ def _upload_product_image(product, image_payload: dict):
product.save(update_fields=['image', 'updated_at'])
@shared_task(bind=True)
def backfill_external_product_images(
self,
merchant_id: int | None = None,
product_id: int | None = None,
limit: int = 500,
dry_run: bool = False,
):
"""
为来自外部印染订单的产品补图。
仅处理:
- printing_order.external_order_id 非空
- product.image 为空
"""
payload = run_external_product_image_backfill(
fetch_image=_fetch_external_product_image,
upload_image=_upload_product_image,
merchant_id=merchant_id,
product_id=product_id,
limit=max(1, int(limit or 500)),
dry_run=bool(dry_run),
on_success=lambda message: logger.info(message),
on_error=lambda message: logger.error(message),
on_info=lambda message: logger.info(message),
)
payload["task_id"] = self.request.id
logger.info("外部印染订单产品补图完成: %s", payload)
return payload
def _record_printing_external_sync_failure(*, run_date, record: dict, error: str):
external_record_id = _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
obj, _created = api_models.PrintingExternalSyncFailure.objects.get_or_create(

View File

@@ -33,6 +33,7 @@ from .views.parameters import StateParameterViewSet
from .views.users import CreateUserWithProfileView
from .views.mingdaoyun import MDYPlateOrderStagingViewSet
from .views.shipment import (
SalesItemDetailView,
SalesItemByCustomerView,
SalesItemByPrintingOrderView,
SalesItemCreateView,
@@ -322,6 +323,11 @@ urlpatterns = [
SalesItemByPrintingOrderView.as_view(),
name="sales_items_by_printing_order",
),
path(
"shipment/sales-items/<int:pk>/",
SalesItemDetailView.as_view(),
name="sales_item_detail",
),
path(
"shipment/sales-items/", SalesItemCreateView.as_view(), name="sales_item_create"
),

View File

@@ -4,6 +4,7 @@ Shipment API 模块
提供出货单和销售品相关的 API 接口
"""
from .views import (
SalesItemDetailView,
SalesItemByCustomerView,
SalesItemByPrintingOrderView,
SalesItemCreateView,
@@ -14,6 +15,7 @@ from .views import (
)
__all__ = [
'SalesItemDetailView',
'SalesItemByCustomerView',
'SalesItemByPrintingOrderView',
'SalesItemCreateView',

View File

@@ -249,6 +249,7 @@ class SalesItemSerializer(serializers.Serializer):
remark = serializers.CharField(read_only=True)
printing_job_id = serializers.IntegerField(read_only=True)
printing_order_id = serializers.SerializerMethodField()
external_order_id = serializers.SerializerMethodField()
customer_id = serializers.IntegerField(read_only=True)
customer_name = serializers.SerializerMethodField()
shipment_id = serializers.IntegerField(
@@ -276,6 +277,16 @@ class SalesItemSerializer(serializers.Serializer):
return printing_job.printing_order_id
return None
def get_external_order_id(self, obj):
external_order_id_map = self.context.get("external_order_id_map", {})
if obj.printing_job_id in external_order_id_map:
return external_order_id_map[obj.printing_job_id]
printing_job = obj.get_printing_job()
if printing_job and getattr(printing_job.printing_order, "external_order_id", None):
return printing_job.printing_order.external_order_id
return None
def get_customer_name(self, obj):
customer_name_map = self.context.get("customer_name_map", {})
if obj.customer_id in customer_name_map:
@@ -293,6 +304,36 @@ class SalesItemSerializer(serializers.Serializer):
return None
class SalesItemDetailSerializer(SalesItemSerializer):
"""
销售品详情序列化器。
在列表字段基础上补充关联印染任务的产品图片。
"""
product_image_url = serializers.SerializerMethodField()
def get_product_image_url(self, obj):
product_image_map = self.context.get("product_image_map", {})
if obj.printing_job_id in product_image_map:
return product_image_map[obj.printing_job_id]
printing_job = obj.get_printing_job()
if not printing_job or not getattr(printing_job, "product", None):
return None
primary_url = printing_job.product.get_primary_image_url()
if primary_url:
return primary_url
if printing_job.product.image:
request = self.context.get("request")
if request:
return request.build_absolute_uri(printing_job.product.image.url)
return printing_job.product.image.url
return None
class ShipmentSalesItemCustomerSerializer(serializers.Serializer):
"""
由未出货销售品反推的客户摘要序列化器。

View File

@@ -81,6 +81,7 @@ class SalesItemByPrintingOrderAPITestCase(TestCase):
width="150cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-PO-001",
)
# 创建印染任务
@@ -216,6 +217,8 @@ class SalesItemByPrintingOrderAPITestCase(TestCase):
self.assertEqual(item["position"], "A1-01")
self.assertEqual(item["remark"], "备注信息")
self.assertEqual(item["printing_job_id"], self.printing_job1.id)
self.assertEqual(item["printing_order_id"], self.printing_order.id)
self.assertEqual(item["external_order_id"], "EXT-PO-001")
self.assertIsNone(item["shipment_id"])
self.assertIsNone(item["shipment_date"])
self.assertIsNotNone(item["created_at"])
@@ -484,6 +487,7 @@ class SalesItemByCustomerAPITestCase(TestCase):
width="150cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-CUST-001",
)
self.printing_job = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
@@ -555,6 +559,7 @@ class SalesItemByCustomerAPITestCase(TestCase):
self.assertEqual(item["customer_name"], self.customer.name)
self.assertEqual(item["printing_job_id"], self.printing_job.id)
self.assertEqual(item["printing_order_id"], self.printing_order.id)
self.assertEqual(item["external_order_id"], "EXT-CUST-001")
self.assertEqual(item["position"], "A1-01")
def test_get_sales_items_by_customer_include_shipped_and_paginate(self):
@@ -571,6 +576,45 @@ class SalesItemByCustomerAPITestCase(TestCase):
result_ids = [item["id"] for item in data["results"]]
self.assertEqual(result_ids, [self.sales_item2.id, self.sales_item_shipped.id])
def test_get_sales_items_by_customer_supports_external_order_id_filter(self):
other_printing_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="其它面料",
width="160cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-CUST-OTHER",
)
other_printing_job = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=other_printing_order,
product=self.product,
quantity=50,
unit="",
created_by=self.user,
)
other_sales_item = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="其它外部订单销售品",
quantity=Decimal("66.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=other_printing_job.id,
customer_id=self.customer.id,
created_by=self.user,
)
resp = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/"
"?external_order_id=EXT-CUST-001"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
result_ids = [item["id"] for item in data["results"]]
self.assertEqual(set(result_ids), {self.sales_item1.id, self.sales_item2.id})
self.assertNotIn(other_sales_item.id, result_ids)
def test_get_sales_items_by_customer_not_found(self):
resp = self.client.get("/api/v1/shipment/sales-items/by-customer/99999/")
@@ -584,6 +628,112 @@ class SalesItemByCustomerAPITestCase(TestCase):
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
class SalesItemDetailAPITestCase(APITestCase):
"""销售品详情 API 测试"""
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.user = User.objects.create_user(
username="sales_item_detail_user",
password="testpass123",
email="sales_item_detail@example.com",
)
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="详情测试员工",
mobile="13800138031",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.other_user = User.objects.create_user(
username="sales_item_detail_other_user",
password="testpass123",
email="sales_item_detail_other@example.com",
)
self.other_employee = basic_models.Employee.objects.create(
sys_user=self.other_user,
merchant=self.other_merchant,
name="其它详情测试员工",
mobile="13800138032",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="详情测试客户",
mobile="13900139033",
area="杭州",
)
self.category = basic_models.ProductCategory.objects.create(
merchant=self.merchant,
name="详情分类",
)
self.product = basic_models.Product.objects.create(
merchant=self.merchant,
category=self.category,
name="详情产品",
human_id="DETAIL001",
mdy_image_url="https://example.com/product-image.jpg",
)
self.state = stateflow_models.State.objects.create(name="详情状态")
self.process = stateflow_models.Process.objects.create(name="详情流程")
self.process.replace_nodes([self.state])
self.printing_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="详情面料",
width="150cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-DETAIL-001",
)
self.printing_job = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=self.printing_order,
product=self.product,
quantity=20,
unit="",
created_by=self.user,
)
self.sales_item = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="详情销售品",
quantity=Decimal("88.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job.id,
customer_id=self.customer.id,
position="B2-03",
created_by=self.user,
)
self.client.force_authenticate(user=self.user)
def test_get_sales_item_detail_success(self):
resp = self.client.get(f"/api/v1/shipment/sales-items/{self.sales_item.id}/")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["id"], self.sales_item.id)
self.assertEqual(data["printing_order_id"], self.printing_order.id)
self.assertEqual(data["external_order_id"], "EXT-DETAIL-001")
self.assertEqual(data["customer_name"], self.customer.name)
self.assertEqual(data["product_image_url"], "https://example.com/product-image.jpg")
def test_get_sales_item_detail_other_merchant_404(self):
self.client.force_authenticate(user=self.other_user)
resp = self.client.get(f"/api/v1/shipment/sales-items/{self.sales_item.id}/")
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
class ShipmentCreateAPITestCase(TestCase):
"""测试创建出货单 API"""

View File

@@ -13,6 +13,7 @@ from flower.viewsets import LimitedLimitOffsetPagination
from shipment.models import Shipment
from .serializers import (
SalesItemDetailSerializer,
SalesItemSerializer,
ShipmentSerializer,
ShipmentCreateNormalSerializer,
@@ -35,18 +36,39 @@ def _build_sales_item_serializer_context(items):
)
printing_order_map = {}
external_order_id_map = {}
product_image_map = {}
if printing_job_ids:
from printing.models import PrintingJob
printing_order_map = dict(
PrintingJob.objects.filter(id__in=printing_job_ids).values_list(
"id", "printing_order_id"
)
printing_jobs = list(
PrintingJob.objects.filter(id__in=printing_job_ids)
.select_related("printing_order", "product")
)
printing_order_map = {
job.id: job.printing_order_id for job in printing_jobs
}
external_order_id_map = {
job.id: getattr(job.printing_order, "external_order_id", None)
for job in printing_jobs
}
for job in printing_jobs:
if not getattr(job, "product", None):
product_image_map[job.id] = None
continue
primary_url = job.product.get_primary_image_url()
if primary_url:
product_image_map[job.id] = primary_url
elif job.product.image:
product_image_map[job.id] = job.product.image.url
else:
product_image_map[job.id] = None
return {
"customer_name_map": customer_name_map,
"printing_order_map": printing_order_map,
"external_order_id_map": external_order_id_map,
"product_image_map": product_image_map,
}
@@ -444,6 +466,7 @@ class SalesItemByCustomerView(GenericAPIView):
request.query_params.get("include_already_has_shipment", "false").lower()
== "true"
)
external_order_id = (request.query_params.get("external_order_id") or "").strip()
from shipment.services import get_sales_items_by_customer
@@ -451,6 +474,7 @@ class SalesItemByCustomerView(GenericAPIView):
merchant=customer.merchant,
customer_id=customer.id,
include_already_has_shipment=include_already_has_shipment,
external_order_id=external_order_id or None,
)
page = self.paginate_queryset(queryset)
items = list(page) if page is not None else list(queryset)
@@ -529,5 +553,48 @@ class SalesItemCreateView(APIView):
return Response({"detail": str(e)}, status=status.HTTP_400_BAD_REQUEST)
# 返回创建的销售品
response_serializer = SalesItemSerializer(sales_item)
response_serializer = SalesItemSerializer(
sales_item,
context=_build_sales_item_serializer_context([sales_item]),
)
return Response(response_serializer.data, status=status.HTTP_201_CREATED)
class SalesItemDetailView(GenericAPIView):
"""
销售品详情。
GET /api/v1/shipment/sales-items/<id>/
"""
permission_classes = [IsAuthenticated]
serializer_class = SalesItemDetailSerializer
def get_queryset(self):
from shipment.models import SalesItem
qs = SalesItem.objects.select_related("shipment", "created_by").order_by("id")
user = self.request.user
if getattr(user, "is_superuser", False):
return qs
emp = getattr(user, "employee", None)
merchant = getattr(emp, "merchant", None) if emp else None
if not merchant:
return SalesItem.objects.none()
return qs.filter(merchant=merchant)
def get(self, request, pk: int):
sales_item = self.get_queryset().filter(id=pk).first()
if sales_item is None:
return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND)
serializer = self.get_serializer(
sales_item,
context={
**_build_sales_item_serializer_context([sales_item]),
"request": request,
},
)
return Response(serializer.data)

View File

@@ -0,0 +1,147 @@
# 外部订单产品补图
本文档说明“外部印染订单产品补图”的手动命令、共享逻辑与定时任务配置。
## 背景
系统中有一类 `PrintingJob` 来自外部印染订单,同步后:
- `printing_order.external_order_id` 有值
- `product.image` 为空
- `job.external_raw.YanSe``job.external_product_name` 中包含可用于外部图片接口查询的名称
为了让产品图片逐步补齐,项目提供:
1. 手动 management command
2. 可复用的共享函数
3. 每小时一次的 Celery 定时任务
## 处理规则
补图任务只处理满足以下条件的数据:
1. `printing_order.external_order_id` 非空
2. `product.image` 为空
3. 能从以下字段提取请求名:
- 优先:`external_raw.YanSe`
- 回退:`external_product_name`
并且同一个 `product` 在单次运行中只会处理一次,避免重复请求外部图片接口。
## 共享逻辑
共享逻辑已抽到:
- [api_v1/external_product_image_backfill.py](/home/f/coding/flower/api_v1/external_product_image_backfill.py)
其中:
- `extract_request_name(job)`
- `run_external_product_image_backfill(...)`
management command 与 Celery task 都复用这套逻辑,避免出现双份实现。
## 手动运行
命令位置:
- [api_v1/management/commands/backfill_external_product_images.py](/home/f/coding/flower/api_v1/management/commands/backfill_external_product_images.py)
### 示例
处理指定商户:
```bash
python manage.py backfill_external_product_images --merchant-id 1
```
限制本次最多处理 50 条:
```bash
python manage.py backfill_external_product_images --merchant-id 1 --limit 50
```
只预览、不实际写库:
```bash
python manage.py backfill_external_product_images --merchant-id 1 --dry-run
```
仅处理指定产品:
```bash
python manage.py backfill_external_product_images --product-id 123
```
### 参数
| 参数 | 说明 |
|------|------|
| `--merchant-id` | 仅处理指定商户的产品 |
| `--product-id` | 仅处理指定产品 |
| `--limit` | 最多处理多少条符合条件的产品 |
| `--dry-run` | 仅输出将处理的产品,不实际请求外部 API 或写库 |
## Celery Task
Task 位置:
- [api_v1/tasks.py](/home/f/coding/flower/api_v1/tasks.py)
Task 名称:
- `api_v1.tasks.backfill_external_product_images`
支持参数:
- `merchant_id`
- `product_id`
- `limit`
- `dry_run`
默认建议:
- `limit=500`
- `dry_run=False`
## 定时任务
已接入 `CELERY_BEAT_SCHEDULE`
- key: `backfill_external_product_images_hourly`
- schedule: `crontab(minute=17)`
- kwargs:
- `limit=500`
- `dry_run=False`
也就是说:
- 每小时第 `17` 分钟运行一次
- 采用错峰策略,避免和若干整点/五分钟任务撞在一起
配置位置:
- [flower/settings.py](/home/f/coding/flower/flower/settings.py)
## 输出与结果
无论是 command 还是 task最终都会产出统一统计结果包含
- `scanned`
- `selected`
- `success`
- `failed`
- `skipped_with_image`
- `skipped_without_request_name`
- `skipped_duplicate_product`
Celery task 还会额外带:
- `task_id`
## 风险与建议
1. 这是外部接口依赖任务,若外部图片接口不稳定,`failed` 可能升高
2. 每小时一次已足够,`limit=500` 对“补漏”场景比较合适
3. 如果后续发现积压较多,可以只调大 `limit`,不必提高调度频率
4. 若未来需要追踪失败明细,可考虑像外部印染同步那样补失败落库;当前版本仅记录日志

View File

@@ -593,6 +593,14 @@ CELERY_BEAT_SCHEDULE = {
'limit': 100,
},
},
'backfill_external_product_images_hourly': {
'task': 'api_v1.tasks.backfill_external_product_images',
'schedule': crontab(minute=17),
'kwargs': {
'limit': 500,
'dry_run': False,
},
},
# 明道云开版:同步到“暂存表”
# - 仅在 02:00-08:00 时间窗内持续运行(每 N 分钟触发一次)
# - 通过 request_interval_seconds 控制单次任务对明道云 API 的请求节奏,避免超 QPS

View File

@@ -0,0 +1,468 @@
import random
from decimal import Decimal
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from basic_info import models as basic_models
from printing import models as printing_models
from shipment import models as shipment_models
class Command(BaseCommand):
help = (
"为开发环境生成一批假的 customer / printing_order / printing_job / sales_item 数据,"
"用于前端联调 shipment 与 printing 相关页面。"
)
def add_arguments(self, parser):
parser.add_argument("--merchant-id", type=int, default=1, help="目标商户 ID默认 1")
parser.add_argument("--user-id", type=int, default=None, help="创建人用户 ID可选")
parser.add_argument("--customers", type=int, default=10, help="客户数量,默认 10")
parser.add_argument("--orders", type=int, default=50, help="订单数量,默认 50")
parser.add_argument(
"--jobs-min", type=int, default=3, help="每个订单最少生成多少个 PrintingJob默认 3"
)
parser.add_argument(
"--jobs-max", type=int, default=30, help="每个订单最多生成多少个 PrintingJob默认 30"
)
parser.add_argument(
"--items-min", type=int, default=2, help="每个 job 最少生成多少个 SalesItem默认 2"
)
parser.add_argument(
"--items-max", type=int, default=20, help="每个 job 最多生成多少个 SalesItem默认 20"
)
parser.add_argument(
"--products", type=int, default=24, help="测试产品数量,默认 24"
)
parser.add_argument(
"--tag",
type=str,
default="DEVSHIP",
help="生成数据的标签前缀,默认 DEVSHIP",
)
parser.add_argument(
"--seed",
type=int,
default=20260331,
help="随机种子,默认 20260331",
)
parser.add_argument(
"--yes",
action="store_true",
help="跳过交互确认,直接执行(谨慎使用)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="仅输出执行前检查与预计写入量,不实际写入数据",
)
def handle(self, *args, **options):
random.seed(options["seed"])
self._validate_options(options)
merchant = basic_models.Merchant.objects.filter(id=options["merchant_id"]).first()
if merchant is None:
raise CommandError(f"merchant_id={options['merchant_id']} 不存在")
created_by = self._resolve_user(merchant=merchant, user_id=options["user_id"])
preview = self._collect_preview(merchant=merchant, created_by=created_by, options=options)
self._print_preview(preview)
if options["dry_run"]:
self.stdout.write(self.style.WARNING("dry-run 模式:未写入任何数据"))
return
if not options["yes"]:
self._confirm_or_abort()
with transaction.atomic():
counts = self._generate_data(
merchant=merchant,
created_by=created_by,
options=options,
)
self._print_summary(
merchant=merchant,
created_by=created_by,
tag=options["tag"],
counts=counts,
)
def _generate_data(self, *, merchant, created_by, options):
category = self._ensure_category(merchant=merchant, tag=options["tag"])
products = self._ensure_products(
merchant=merchant,
category=category,
count=options["products"],
tag=options["tag"],
)
customers = self._create_customers(
merchant=merchant,
count=options["customers"],
tag=options["tag"],
)
order_count = 0
job_count = 0
sales_item_count = 0
self.stdout.write(self.style.NOTICE("开始生成测试数据..."))
self.stdout.write("步骤 1/3: 准备客户与产品")
self.stdout.write(f" 客户数: {len(customers)}")
self.stdout.write(f" 产品数: {len(products)}")
self.stdout.write("步骤 2/3: 创建 PrintingOrder / PrintingJob / SalesItem")
for index in range(1, options["orders"] + 1):
customer = customers[(index - 1) % len(customers)]
order = printing_models.PrintingOrder.objects.create(
merchant=merchant,
customer=customer,
fabric=random.choice(["全棉", "涤纶", "尼龙", "人棉", "TC"]),
width=random.choice(["150cm", "160cm", "170cm", "180cm"]),
is_urgent=random.choice([False, False, False, True]),
area=customer.area or random.choice(["广州", "佛山", "绍兴", "杭州"]),
address=f"{customer.area or '广州'}-{index}号测试地址",
fabric_source=random.choice(["自带布", "仓库", "客户送布"]),
is_fabric_received=random.choice([True, False]),
craft=random.choice(["活性印花", "数码印花", "涂料印花"]),
description=f"{options['tag']} 测试印染订单 #{index}",
position=random.choice(["A区", "B区", "C区", "待排产"]),
printing_warn=random.choice(["", "注意色差", "先确认花型"]),
rolling_warn=random.choice(["", "注意卷边", "慢速滚筒"]),
production_warn=random.choice(["", "先打样", "优先安排"]),
created_by=created_by,
external_order_id=self._build_external_order_id(index=index),
external_customer_id=f"CUST-{customer.id}",
external_customer_name=customer.name,
external_employee_name=self._display_user(created_by),
)
order_count += 1
jobs_for_order = random.randint(options["jobs_min"], options["jobs_max"])
for job_index in range(1, jobs_for_order + 1):
product = random.choice(products)
job = printing_models.PrintingJob.objects.create(
merchant=merchant,
printing_order=order,
product=product,
work_state=random.choice(
[
printing_models.PrintingJobWorkStateEnum.PRODUCING,
printing_models.PrintingJobWorkStateEnum.WAITING_FOR_DELIVERY,
printing_models.PrintingJobWorkStateEnum.WAITING_FOR_INVOICE,
]
),
quantity=random.randint(80, 3000),
unit=random.choice(["", "", ""]),
size=random.choice(["", "120x120", "150x150", "200x200"]) or None,
pieces=random.choice([None, 1, 2, 4, 6, 8]),
description=f"{options['tag']} Job {index}-{job_index}",
created_by=created_by,
external_product_name=product.name,
)
job_count += 1
sales_items = []
items_for_job = random.randint(options["items_min"], options["items_max"])
for item_index in range(1, items_for_job + 1):
sales_items.append(
shipment_models.SalesItem(
shipment=None,
merchant=merchant,
name=f"{product.name}-销售品-{index}-{job_index}-{item_index}",
quantity=Decimal(
f"{random.randint(20, 3000)}.{random.randint(0, 99):02d}"
),
unit=random.choice(
[
shipment_models.UnitChoices.METER,
shipment_models.UnitChoices.PIECE,
shipment_models.UnitChoices.YARD,
shipment_models.UnitChoices.UNIT,
]
),
created_by=created_by,
printing_job_id=job.id,
customer_id=customer.id,
position=random.choice(
[
"A1-01",
"A1-02",
"B2-03",
"C3-05",
"待分区",
"待发货区",
]
),
remark=random.choice(["", "测试数据", "扫码联调用", "优先出货"]),
)
)
shipment_models.SalesItem.objects.bulk_create(sales_items, batch_size=500)
sales_item_count += len(sales_items)
if index % 10 == 0 or index == options["orders"]:
self.stdout.write(
f" 已完成订单 {index}/{options['orders']}"
f"累计 jobs={job_count}sales_items={sales_item_count}"
)
self.stdout.write("步骤 3/3: 数据写入完成")
return {
"customers_created": len(customers),
"printing_orders_created": order_count,
"printing_jobs_created": job_count,
"sales_items_created": sales_item_count,
}
def _resolve_user(self, *, merchant, user_id):
User = get_user_model()
if user_id is not None:
user = User.objects.filter(id=user_id).first()
if user is None:
raise CommandError(f"user_id={user_id} 不存在")
employee = getattr(user, "employee", None)
if employee is None or employee.merchant_id != merchant.id:
raise CommandError("指定 user 未绑定到目标商户的 employee")
return user
employee = (
basic_models.Employee.objects.select_related("sys_user")
.filter(merchant=merchant, sys_user__isnull=False)
.order_by("id")
.first()
)
if employee and employee.sys_user:
return employee.sys_user
username = f"seed_{merchant.id}_user"
user = User.objects.create_user(username=username, password="devpass123")
basic_models.Employee.objects.create(
merchant=merchant,
sys_user=user,
name=f"Seed User {merchant.id}",
mobile=f"1390000{merchant.id:04d}",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
return user
def _ensure_category(self, *, merchant, tag):
category, _ = basic_models.ProductCategory.objects.get_or_create(
merchant=merchant,
name=f"{tag}-测试分类",
defaults={"product_prefix": "TS"},
)
return category
def _ensure_products(self, *, merchant, category, count, tag):
products = list(
basic_models.Product.objects.filter(
merchant=merchant,
category=category,
name__startswith=f"{tag}-测试产品-",
).order_by("id")
)
missing = count - len(products)
for index in range(1, missing + 1):
number = len(products) + index
products.append(
basic_models.Product.objects.create(
merchant=merchant,
category=category,
name=f"{tag}-测试产品-{number:03d}",
human_id=f"{tag[:4]}P{number:03d}",
unit=random.choice(
[
basic_models.ProductUnitEnum.METER,
basic_models.ProductUnitEnum.YARD,
basic_models.ProductUnitEnum.SEGMENT,
]
),
color=random.choice(["", "", "", "", ""]),
width_size=Decimal(random.choice(["150.00", "160.00", "170.00"])),
description="开发环境自动生成的测试产品",
from_mdy=False,
)
)
return list(
basic_models.Product.objects.filter(
merchant=merchant,
category=category,
name__startswith=f"{tag}-测试产品-",
).order_by("id")[:count]
)
def _create_customers(self, *, merchant, count, tag):
customers = []
employee = (
basic_models.Employee.objects.filter(merchant=merchant).order_by("id").first()
)
if employee is None:
raise CommandError("目标商户下没有 employee无法创建 customer")
for index in range(1, count + 1):
name = f"{tag}-测试客户-{index:02d}"
customer, _ = basic_models.Customer.objects.get_or_create(
merchant=merchant,
name=name,
defaults={
"created_by": employee,
"mobile": f"138{merchant.id:02d}{index:06d}"[:11],
"area": random.choice(["广州", "佛山", "绍兴", "杭州", "苏州"]),
"contact": f"联系人{index:02d}",
"description": "开发环境自动生成的测试客户",
},
)
customers.append(customer)
return customers
def _build_external_order_id(self, *, index):
return f"KD{20135142 + index:08d}"
def _validate_options(self, options):
if options["customers"] <= 0:
raise CommandError("--customers 必须大于 0")
if options["orders"] <= 0:
raise CommandError("--orders 必须大于 0")
if options["products"] <= 0:
raise CommandError("--products 必须大于 0")
if options["jobs_min"] <= 0 or options["jobs_max"] <= 0:
raise CommandError("--jobs-min / --jobs-max 必须大于 0")
if options["items_min"] <= 0 or options["items_max"] <= 0:
raise CommandError("--items-min / --items-max 必须大于 0")
if options["jobs_min"] > options["jobs_max"]:
raise CommandError("--jobs-min 不能大于 --jobs-max")
if options["items_min"] > options["items_max"]:
raise CommandError("--items-min 不能大于 --items-max")
def _collect_preview(self, *, merchant, created_by, options):
db = settings.DATABASES["default"]
tag = options["tag"]
existing_customer_count = basic_models.Customer.objects.filter(
merchant=merchant,
name__startswith=f"{tag}-测试客户-",
).count()
existing_order_count = printing_models.PrintingOrder.objects.filter(
merchant=merchant,
description__startswith=f"{tag} 测试印染订单 #",
).count()
existing_job_count = printing_models.PrintingJob.objects.filter(
merchant=merchant,
description__startswith=f"{tag} Job ",
).count()
existing_sales_item_count = shipment_models.SalesItem.objects.filter(
merchant=merchant,
name__startswith=f"{tag}-测试产品-",
).count()
jobs_min_total = options["orders"] * options["jobs_min"]
jobs_max_total = options["orders"] * options["jobs_max"]
sales_items_min_total = jobs_min_total * options["items_min"]
sales_items_max_total = jobs_max_total * options["items_max"]
return {
"debug": getattr(settings, "DEBUG", None),
"db_host": db.get("HOST"),
"db_port": db.get("PORT"),
"db_name": db.get("NAME"),
"merchant_id": merchant.id,
"merchant_name": merchant.name,
"created_by_id": created_by.id,
"created_by_username": created_by.username,
"tag": tag,
"seed": options["seed"],
"customers": options["customers"],
"orders": options["orders"],
"jobs_min": options["jobs_min"],
"jobs_max": options["jobs_max"],
"items_min": options["items_min"],
"items_max": options["items_max"],
"products": options["products"],
"existing_customer_count": existing_customer_count,
"existing_order_count": existing_order_count,
"existing_job_count": existing_job_count,
"existing_sales_item_count": existing_sales_item_count,
"jobs_min_total": jobs_min_total,
"jobs_max_total": jobs_max_total,
"sales_items_min_total": sales_items_min_total,
"sales_items_max_total": sales_items_max_total,
}
def _print_preview(self, preview):
self.stdout.write(self.style.WARNING("即将生成开发环境测试数据,请先确认以下状态"))
self.stdout.write("=" * 72)
self.stdout.write(f"DEBUG : {preview['debug']}")
self.stdout.write(
f"数据库 : {preview['db_name']} @ {preview['db_host']}:{preview['db_port']}"
)
self.stdout.write(
f"目标商户 : {preview['merchant_id']} / {preview['merchant_name']}"
)
self.stdout.write(
f"创建人用户 : {preview['created_by_id']} / {preview['created_by_username']}"
)
self.stdout.write(f"数据标签 : {preview['tag']}")
self.stdout.write(f"随机种子 : {preview['seed']}")
self.stdout.write("-" * 72)
self.stdout.write(f"客户数 : {preview['customers']}")
self.stdout.write(f"订单数 : {preview['orders']}")
self.stdout.write(
f"每单 job 数 : {preview['jobs_min']} ~ {preview['jobs_max']}"
)
self.stdout.write(
f"每 job sales item 数: {preview['items_min']} ~ {preview['items_max']}"
)
self.stdout.write(f"测试产品数 : {preview['products']}")
self.stdout.write("-" * 72)
self.stdout.write(
f"预计新增 PrintingJob: {preview['jobs_min_total']} ~ {preview['jobs_max_total']}"
)
self.stdout.write(
"预计新增 SalesItem : "
f"{preview['sales_items_min_total']} ~ {preview['sales_items_max_total']}"
)
self.stdout.write("-" * 72)
self.stdout.write(
f"当前同 tag 客户数 : {preview['existing_customer_count']}"
)
self.stdout.write(
f"当前同 tag 订单数 : {preview['existing_order_count']}"
)
self.stdout.write(
f"当前同 tag job 数 : {preview['existing_job_count']}"
)
self.stdout.write(
f"当前同 tag sales item: {preview['existing_sales_item_count']}"
)
self.stdout.write("=" * 72)
def _confirm_or_abort(self):
answer = input("确认要继续写入这批测试数据吗?请输入 YES 继续: ").strip()
if answer != "YES":
raise CommandError("已取消执行,未写入任何数据")
def _print_summary(self, *, merchant, created_by, tag, counts):
self.stdout.write(self.style.SUCCESS("本次执行摘要"))
self.stdout.write("=" * 72)
self.stdout.write(f"merchant_id : {merchant.id}")
self.stdout.write(f"merchant_name : {merchant.name}")
self.stdout.write(f"created_by : {created_by.id}:{created_by.username}")
self.stdout.write(f"tag : {tag}")
self.stdout.write(f"customers_created : {counts['customers_created']}")
self.stdout.write(f"printing_orders : {counts['printing_orders_created']}")
self.stdout.write(f"printing_jobs : {counts['printing_jobs_created']}")
self.stdout.write(f"sales_items : {counts['sales_items_created']}")
self.stdout.write("=" * 72)
self.stdout.write(self.style.SUCCESS("假数据生成完成"))
def _display_user(self, user):
employee = getattr(user, "employee", None)
if employee:
return employee.name
return user.username

View File

@@ -83,6 +83,7 @@ def get_sales_items_by_customer(
merchant,
customer_id: int,
include_already_has_shipment: bool = False,
external_order_id: str | None = None,
) -> QuerySet[SalesItem]:
"""
通过客户ID查询对应销售品。
@@ -94,6 +95,15 @@ def get_sales_items_by_customer(
customer_id=customer_id,
)
if external_order_id:
from printing.models import PrintingJob
printing_job_ids = PrintingJob.objects.filter(
merchant=merchant,
printing_order__external_order_id=external_order_id,
).values_list("id", flat=True)
queryset = queryset.filter(printing_job_id__in=list(printing_job_ids))
if not include_already_has_shipment:
queryset = queryset.filter(shipment__isnull=True)