feat: added mdy_product_sync new task for merchant_id isolated

api_v1/mdy_plate_order_sync.py

@@ -135,6 +135,7 @@ def sync_mdy_plate_orders_to_staging(
     use_checkpoint: bool = True,
     update_checkpoint: bool = True,
     sort_direction: Literal["asc", "desc"] = "asc",
+    merchant_id: int | None = None,
 ) -> dict:
     """同步明道云开版数据表到暂存表（含可选跨表关联数据）。
 
@@ -161,7 +162,8 @@ def sync_mdy_plate_orders_to_staging(
     if use_checkpoint:
         last_sync = (
             api_models.DataSync.objects.filter(
-                table_name=api_models.DataSync.TableName.PLATE_ORDER
+                table_name=api_models.DataSync.TableName.PLATE_ORDER,
+                merchant_id=merchant_id,
             )
             .order_by("-created_at")
             .first()
@@ -292,6 +294,7 @@ def sync_mdy_plate_orders_to_staging(
     if update_checkpoint:
         api_models.DataSync.objects.create(
             table_name=api_models.DataSync.TableName.PLATE_ORDER,
+            merchant_id=merchant_id,
             page_index=page_index,
             page_size=page_size,
             synced_rows=synced_rows,
@@ -315,5 +318,5 @@ def sync_mdy_plate_orders_to_staging(
         "use_checkpoint": use_checkpoint,
         "update_checkpoint": update_checkpoint,
     }
-    logger.info("明道云开版暂存同步完成: %s", payload)
+    logger.info("明道云开版暂存同步完成(merchant=%s): %s", merchant_id, payload)
     return payload

api_v1/migrations/0012_datasync_merchant.py

@@ -0,0 +1,25 @@
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('basic_info', '0024_frontend_page_and_visible_pages'),
+        ('api_v1', '0011_alter_datasync_table_name'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='datasync',
+            name='merchant',
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name='data_sync_records',
+                to='basic_info.merchant',
+                verbose_name='所属商户',
+            ),
+        ),
+    ]

api_v1/models.py

@@ -7,6 +7,7 @@ from django.db import models
 from django.contrib.auth import get_user_model
 from django.db.models.fields.json import KeyTextTransform
 from flower.common import ModelBase
+from basic_info import models as basic_info_models
 
 User = get_user_model()
 
@@ -95,6 +96,14 @@ class DataSync(ModelBase):
         MDY_PLATE_ORDER_STAGING_TIIA_UPLOAD = 'mdy_plate_order_staging_tiia_upload', '开版暂存-腾讯图库上传'
 
     table_name = models.CharField(max_length=50, choices=TableName.choices, verbose_name='同步目标')
+    merchant = models.ForeignKey(
+        basic_info_models.Merchant,
+        on_delete=models.PROTECT,
+        null=True,
+        blank=True,
+        related_name='data_sync_records',
+        verbose_name='所属商户',
+    )
     page_index = models.PositiveIntegerField(default=1, verbose_name='页码')
     page_size = models.PositiveIntegerField(default=100, verbose_name='每页数量')
     total_count = models.PositiveIntegerField(default=0, verbose_name='远端总数')

api_v1/tasks.py

@@ -103,8 +103,8 @@ def backup_database(self, output_dir: str | None = None, filename_prefix: str =
     return payload
 
 
-def _get_mdy_merchant():
+def _get_mdy_merchant(merchant_id: int | None = None):
-    merchant_id = getattr(settings, 'MDY_MERCHANT_ID', None)
+    merchant_id = merchant_id or getattr(settings, 'MDY_MERCHANT_ID', None)
     qs = basic_models.Merchant.objects.all()
     if merchant_id:
         qs = qs.filter(id=merchant_id)
@@ -230,7 +230,12 @@ def _upsert_product(product_data, merchant, category):
 
 
 @shared_task(bind=True)
-def sync_mdy_products(self, page_size: int = 300, max_pages: int | None = None, max_records: int | None = None):
+def sync_mdy_products(
+    self,
+    page_size: int = 300,
+    max_pages: int | None = None,
+    max_records: int | None = None,
+):
     """
     从明道云同步产品数据（按 ctime 升序分页扫描）。
 
@@ -243,7 +248,8 @@ def sync_mdy_products(self, page_size: int = 300, max_pages: int | None = None,
     max_records = max_records or 0  # 0 表示不限制
 
     last_sync = api_models.DataSync.objects.filter(
-        table_name=api_models.DataSync.TableName.PRODUCT
+        table_name=api_models.DataSync.TableName.PRODUCT,
+        merchant__isnull=True,
     ).order_by('-created_at').first()
     last_ctime = last_sync.last_ctime if last_sync else None
     last_rowid = last_sync.last_rowid if last_sync else ''
@@ -307,6 +313,7 @@ def sync_mdy_products(self, page_size: int = 300, max_pages: int | None = None,
     record_last_rowid = latest_rowid or last_rowid
     api_models.DataSync.objects.create(
         table_name=api_models.DataSync.TableName.PRODUCT,
+        merchant=None,
         page_index=page_index,
         page_size=page_size,
         synced_rows=synced_rows,
@@ -362,7 +369,12 @@ def _upsert_customer(customer_data, merchant):
 
 
 @shared_task(bind=True)
-def sync_mdy_customers(self, page_size: int = 300, max_pages: int | None = None, max_records: int | None = None):
+def sync_mdy_customers(
+    self,
+    page_size: int = 300,
+    max_pages: int | None = None,
+    max_records: int | None = None,
+):
     """
     从明道云同步客户数据（按 ctime 升序分页扫描）。
 
@@ -374,7 +386,8 @@ def sync_mdy_customers(self, page_size: int = 300, max_pages: int | None = None,
     max_records = max_records or 0  # 0 表示不限制
 
     last_sync = api_models.DataSync.objects.filter(
-        table_name=api_models.DataSync.TableName.CUSTOMER
+        table_name=api_models.DataSync.TableName.CUSTOMER,
+        merchant__isnull=True,
     ).order_by('-created_at').first()
     last_ctime = last_sync.last_ctime if last_sync else None
     last_rowid = last_sync.last_rowid if last_sync else ''
@@ -434,6 +447,7 @@ def sync_mdy_customers(self, page_size: int = 300, max_pages: int | None = None,
     record_last_rowid = latest_rowid or last_rowid
     api_models.DataSync.objects.create(
         table_name=api_models.DataSync.TableName.CUSTOMER,
+        merchant=None,
         page_index=page_index,
         page_size=page_size,
         synced_rows=synced_rows,
@@ -474,11 +488,234 @@ def sync_mdy_plate_orders(
         with_related=with_related,
         max_related_per_type=max_related_per_type,
         request_interval_seconds=request_interval_seconds,
+        merchant_id=None,
     )
     payload["task_id"] = self.request.id
     return payload
 
 
+@shared_task(bind=True)
+def sync_mdy_products_for_merchant(
+    self,
+    *,
+    merchant_id: int,
+    page_size: int = 300,
+    max_pages: int | None = None,
+    max_records: int | None = None,
+):
+    """按商户隔离同步产品数据（独立游标）。"""
+    merchant = _get_mdy_merchant(merchant_id=merchant_id)
+    category = _get_mdy_category(merchant)
+    max_records = max_records or 0
+
+    last_sync = api_models.DataSync.objects.filter(
+        table_name=api_models.DataSync.TableName.PRODUCT,
+        merchant=merchant,
+    ).order_by('-created_at').first()
+    last_ctime = last_sync.last_ctime if last_sync else None
+    last_rowid = last_sync.last_rowid if last_sync else ''
+    start_page_index = last_sync.page_index if last_sync else 1
+
+    synced_rows = 0
+    page_index = max(1, start_page_index)
+    pages_processed = 0
+    total_count = 0
+    latest_ctime = last_ctime
+    latest_rowid = last_rowid
+
+    while True:
+        if max_pages is not None and pages_processed >= max_pages:
+            break
+        if max_records and synced_rows >= max_records:
+            break
+
+        products, total = _run_fetch(page_index, page_size)
+        total_count = total
+        if not products:
+            break
+
+        hit_max_records = False
+        for item in products:
+            if max_records and synced_rows >= max_records:
+                hit_max_records = True
+                break
+            product_ctime = _parse_mdy_datetime(item.created_at)
+
+            if last_ctime and product_ctime:
+                if product_ctime < last_ctime:
+                    continue
+                if product_ctime == last_ctime and last_rowid and item.rowid == last_rowid:
+                    continue
+
+            changed = _upsert_product(item, merchant, category)
+            if changed:
+                synced_rows += 1
+            if product_ctime:
+                if latest_ctime is None or product_ctime > latest_ctime:
+                    latest_ctime = product_ctime
+                    latest_rowid = item.rowid
+                elif product_ctime == latest_ctime:
+                    latest_rowid = item.rowid
+
+        pages_processed += 1
+        if hit_max_records:
+            break
+
+        if len(products) < page_size:
+            break
+        page_index += 1
+
+    record_last_ctime = latest_ctime or last_ctime
+    record_last_rowid = latest_rowid or last_rowid
+    api_models.DataSync.objects.create(
+        table_name=api_models.DataSync.TableName.PRODUCT,
+        merchant=merchant,
+        page_index=page_index,
+        page_size=page_size,
+        synced_rows=synced_rows,
+        total_count=total_count,
+        last_ctime=record_last_ctime,
+        last_rowid=record_last_rowid,
+        note='asc scan (merchant isolated)',
+    )
+
+    payload = {
+        'task_id': self.request.id,
+        'merchant_id': merchant.id,
+        'synced_rows': synced_rows,
+        'page_index': page_index,
+        'page_size': page_size,
+        'total_count': total_count,
+        'last_ctime': record_last_ctime.isoformat() if record_last_ctime else None,
+    }
+    logger.info('明道云产品同步完成(merchant=%s): %s', merchant.id, payload)
+    return payload
+
+
+@shared_task(bind=True)
+def sync_mdy_customers_for_merchant(
+    self,
+    *,
+    merchant_id: int,
+    page_size: int = 300,
+    max_pages: int | None = None,
+    max_records: int | None = None,
+):
+    """按商户隔离同步客户数据（独立游标）。"""
+    merchant = _get_mdy_merchant(merchant_id=merchant_id)
+    max_records = max_records or 0
+
+    last_sync = api_models.DataSync.objects.filter(
+        table_name=api_models.DataSync.TableName.CUSTOMER,
+        merchant=merchant,
+    ).order_by('-created_at').first()
+    last_ctime = last_sync.last_ctime if last_sync else None
+    last_rowid = last_sync.last_rowid if last_sync else ''
+    start_page_index = last_sync.page_index if last_sync else 1
+
+    synced_rows = 0
+    page_index = max(1, start_page_index)
+    pages_processed = 0
+    total_count = 0
+    latest_ctime = last_ctime
+    latest_rowid = last_rowid
+
+    while True:
+        if max_pages is not None and pages_processed >= max_pages:
+            break
+        if max_records and synced_rows >= max_records:
+            break
+
+        customers, total = _run_fetch_customers(page_index, page_size)
+        total_count = total
+        if not customers:
+            break
+
+        hit_max_records = False
+        for item in customers:
+            if max_records and synced_rows >= max_records:
+                hit_max_records = True
+                break
+            record_ctime = _parse_mdy_datetime(item.created_at)
+
+            if last_ctime and record_ctime:
+                if record_ctime < last_ctime:
+                    continue
+                if record_ctime == last_ctime and last_rowid and item.rowid == last_rowid:
+                    continue
+
+            changed = _upsert_customer(item, merchant)
+            if changed:
+                synced_rows += 1
+            if record_ctime:
+                if latest_ctime is None or record_ctime > latest_ctime:
+                    latest_ctime = record_ctime
+                    latest_rowid = item.rowid
+                elif record_ctime == latest_ctime:
+                    latest_rowid = item.rowid
+
+        pages_processed += 1
+        if hit_max_records:
+            break
+
+        if len(customers) < page_size:
+            break
+        page_index += 1
+
+    record_last_ctime = latest_ctime or last_ctime
+    record_last_rowid = latest_rowid or last_rowid
+    api_models.DataSync.objects.create(
+        table_name=api_models.DataSync.TableName.CUSTOMER,
+        merchant=merchant,
+        page_index=page_index,
+        page_size=page_size,
+        synced_rows=synced_rows,
+        total_count=total_count,
+        last_ctime=record_last_ctime,
+        last_rowid=record_last_rowid,
+        note='asc scan (merchant isolated)',
+    )
+
+    payload = {
+        'task_id': self.request.id,
+        'merchant_id': merchant.id,
+        'synced_rows': synced_rows,
+        'page_index': page_index,
+        'page_size': page_size,
+        'total_count': total_count,
+        'last_ctime': record_last_ctime.isoformat() if record_last_ctime else None,
+    }
+    logger.info('明道云客户同步完成(merchant=%s): %s', merchant.id, payload)
+    return payload
+
+
+@shared_task(bind=True)
+def sync_mdy_plate_orders_for_merchant(
+    self,
+    *,
+    merchant_id: int,
+    page_size: int = 300,
+    max_pages: int | None = None,
+    max_records: int | None = None,
+    with_related: bool = True,
+    max_related_per_type: int = 5,
+    request_interval_seconds: float = 0.02,
+):
+    """按商户隔离同步开版暂存表（独立游标）。"""
+    payload = sync_mdy_plate_orders_to_staging(
+        page_size=page_size,
+        max_pages=max_pages,
+        max_records=max_records,
+        with_related=with_related,
+        max_related_per_type=max_related_per_type,
+        request_interval_seconds=request_interval_seconds,
+        merchant_id=merchant_id,
+    )
+    payload["task_id"] = self.request.id
+    payload["merchant_id"] = merchant_id
+    return payload
+
+
 def _record_mdy_plate_order_staging_tiia_failure(
     *,
     run_date,

flower/settings.py

@@ -60,6 +60,7 @@ CSRF_TRUSTED_ORIGINS = env.list('CSRF_TRUSTED_ORIGINS', default=[
 # 明道云同步配置
 MDY_MERCHANT_ID = env.int('MDY_MERCHANT_ID', default=1)
 MDY_PRODUCT_CATEGORY_ID = env.int('MDY_PRODUCT_CATEGORY_ID', default=1)
+MDY_ISOLATED_MERCHANT_ID = env.int('MDY_ISOLATED_MERCHANT_ID', default=0)
 MDY_SYNC_PAGE_SIZE = env.int('MDY_SYNC_PAGE_SIZE', default=100)
 MDY_SYNC_MAX_PAGES = env.int('MDY_SYNC_MAX_PAGES', default=2)
 MDY_SYNC_MAX_RECORDS = env.int('MDY_SYNC_MAX_RECORDS', default=200)
@@ -581,3 +582,42 @@ CELERY_BEAT_SCHEDULE = {
         },
     },
 }
+
+if MDY_ISOLATED_MERCHANT_ID and MDY_ISOLATED_MERCHANT_ID > 0:
+    CELERY_BEAT_SCHEDULE.update(
+        {
+            'mdy_product_sync_isolated': {
+                'task': 'api_v1.tasks.sync_mdy_products_for_merchant',
+                'schedule': crontab(minute='3,13,23,33,43,53'),
+                'kwargs': {
+                    'merchant_id': MDY_ISOLATED_MERCHANT_ID,
+                    'page_size': MDY_SYNC_PAGE_SIZE,
+                    'max_pages': MDY_SYNC_MAX_PAGES,
+                    'max_records': MDY_SYNC_MAX_RECORDS,
+                },
+            },
+            # 'mdy_customer_sync_isolated': {
+            #     'task': 'api_v1.tasks.sync_mdy_customers_for_merchant',
+            #     'schedule': crontab(minute='1,11,21,31,41,51'),
+            #     'kwargs': {
+            #         'merchant_id': MDY_ISOLATED_MERCHANT_ID,
+            #         'page_size': MDY_SYNC_PAGE_SIZE,
+            #         'max_pages': MDY_SYNC_MAX_PAGES,
+            #         'max_records': MDY_SYNC_MAX_RECORDS,
+            #     },
+            # },
+            # 'mdy_plate_order_staging_sync_isolated': {
+            #     'task': 'api_v1.tasks.sync_mdy_plate_orders_for_merchant',
+            #     'schedule': crontab(minute='*/2', hour='8-11'),
+            #     'kwargs': {
+            #         'merchant_id': MDY_ISOLATED_MERCHANT_ID,
+            #         'page_size': max(1, int(MDY_SYNC_PAGE_SIZE)),
+            #         'max_pages': max(1, int(MDY_SYNC_MAX_PAGES)),
+            #         'max_records': max(1, int(MDY_SYNC_MAX_RECORDS)),
+            #         'with_related': True,
+            #         'max_related_per_type': 5,
+            #         'request_interval_seconds': float(MDY_PLATE_ORDER_SYNC_REQUEST_INTERVAL_SECONDS),
+            #     },
+            # },
+        }
+    )