From 0e56412f5dd68126eaf7ad02b81e070534c5b3ca Mon Sep 17 00:00:00 2001 From: colaftc Date: Wed, 4 Feb 2026 18:47:07 +0800 Subject: [PATCH] feat: added mdy_product_sync new task for merchant_id isolated --- api_v1/mdy_plate_order_sync.py | 7 +- api_v1/migrations/0012_datasync_merchant.py | 25 ++ api_v1/models.py | 9 + api_v1/tasks.py | 249 +++++++++++++++++++- flower/settings.py | 40 ++++ 5 files changed, 322 insertions(+), 8 deletions(-) create mode 100644 api_v1/migrations/0012_datasync_merchant.py diff --git a/api_v1/mdy_plate_order_sync.py b/api_v1/mdy_plate_order_sync.py index 6ecadb1..5a80e07 100644 --- a/api_v1/mdy_plate_order_sync.py +++ b/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 diff --git a/api_v1/migrations/0012_datasync_merchant.py b/api_v1/migrations/0012_datasync_merchant.py new file mode 100644 index 0000000..c5904fc --- /dev/null +++ b/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='所属商户', + ), + ), + ] diff --git a/api_v1/models.py b/api_v1/models.py index 28b3771..8e0dd6b 100644 --- a/api_v1/models.py +++ b/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='远端总数') diff --git a/api_v1/tasks.py b/api_v1/tasks.py index afac0ad..9074fd7 100644 --- a/api_v1/tasks.py +++ b/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(): - merchant_id = getattr(settings, 'MDY_MERCHANT_ID', None) +def _get_mdy_merchant(merchant_id: int | None = 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, diff --git a/flower/settings.py b/flower/settings.py index cd00231..7ab3fcf 100644 --- a/flower/settings.py +++ b/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), + # }, + # }, + } + )