feat: refactor flower.utils to sub package, added some sync api in package, added plate_order sync result model but relation of plate_order was not query (do it next step)

api_v1/tasks.py

@@ -220,7 +220,11 @@ 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):
     """
-    从明道云同步产品数据（按 ctime 升序遍历，依赖 last_ctime/rowid 游标）
+    从明道云同步产品数据（按 ctime 升序分页扫描）。
+
+    - 默认从上次同步记录的 page_index 继续翻页
+    - last_ctime/last_rowid 用于页内游标（避免重复处理）
+    - max_pages 表示“单次任务最多处理多少页”（不是最大页码）
     """
     merchant = _get_mdy_merchant()
     category = _get_mdy_category(merchant)
@@ -231,15 +235,18 @@ def sync_mdy_products(self, page_size: int = 300, max_pages: int | None = None,
     ).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 = 1
+    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 page_index > max_pages:
+        # max_pages: 单次任务最多处理多少页（不是“最大页码”）
+        if max_pages is not None and pages_processed >= max_pages:
             break
         if max_records and synced_rows >= max_records:
             break
@@ -249,8 +256,10 @@ def sync_mdy_products(self, page_size: int = 300, max_pages: int | None = None,
         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)
 
@@ -264,9 +273,18 @@ def sync_mdy_products(self, page_size: int = 300, max_pages: int | None = None,
             changed = _upsert_product(item, merchant, category)
             if changed:
                 synced_rows += 1
-            if product_ctime and (latest_ctime is None or product_ctime > latest_ctime):
-                latest_ctime = product_ctime
-                latest_rowid = item.rowid
+            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:
+                    # 同一秒内可能有多条记录，尽量把 rowid 推进到最后处理的那条
+                    latest_rowid = item.rowid
+
+        pages_processed += 1
+        if hit_max_records:
+            # 达到单次任务的记录上限：下次从同一页继续（依赖 last_ctime/last_rowid 跳过已处理部分）
+            break
 
         # 若返回不足一页，说明到尾部，可结束
         if len(products) < page_size:
@@ -334,7 +352,11 @@ 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):
     """
-    从明道云同步客户数据（按 ctime 升序遍历，依赖 last_ctime/rowid 游标）
+    从明道云同步客户数据（按 ctime 升序分页扫描）。
+
+    - 默认从上次同步记录的 page_index 继续翻页
+    - last_ctime/last_rowid 用于页内游标（避免重复处理）
+    - max_pages 表示“单次任务最多处理多少页”（不是最大页码）
     """
     merchant = _get_mdy_merchant()
     max_records = max_records or 0  # 0 表示不限制
@@ -344,15 +366,18 @@ def sync_mdy_customers(self, page_size: int = 300, max_pages: int | None = None,
     ).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 = 1
+    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 page_index > max_pages:
+        # max_pages: 单次任务最多处理多少页（不是“最大页码”）
+        if max_pages is not None and pages_processed >= max_pages:
             break
         if max_records and synced_rows >= max_records:
             break
@@ -362,8 +387,10 @@ def sync_mdy_customers(self, page_size: int = 300, max_pages: int | None = None,
         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)
 
@@ -376,9 +403,16 @@ def sync_mdy_customers(self, page_size: int = 300, max_pages: int | None = None,
             changed = _upsert_customer(item, merchant)
             if changed:
                 synced_rows += 1
-            if record_ctime and (latest_ctime is None or record_ctime > latest_ctime):
-                latest_ctime = record_ctime
-                latest_rowid = item.rowid
+            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