From a36a7e5265e8867a4a85b28653588d6ec0a2acdc Mon Sep 17 00:00:00 2001 From: colaftc Date: Mon, 8 Dec 2025 22:22:18 +0800 Subject: [PATCH] feat: tasks for sync products and customers --- api_v1/admin.py | 12 +- api_v1/migrations/0003_datasync.py | 33 ++ api_v1/models.py | 26 + api_v1/tasks.py | 330 ++++++++++-- api_v1/views/printing/views.py | 2 + basic_info/admin.py | 2 +- .../migrations/0018_product_from_mdy.py | 20 + .../migrations/0019_customer_mdy_fields.py | 21 + basic_info/models.py | 18 + celerybeat-schedule | Bin 12288 -> 12288 bytes celerybeat-schedule-shm | Bin 32768 -> 32768 bytes celerybeat-schedule-wal | Bin 444992 -> 57712 bytes flower/settings.py | 25 + flower/utils.py | 476 ++++++++++-------- 14 files changed, 717 insertions(+), 248 deletions(-) create mode 100644 api_v1/migrations/0003_datasync.py create mode 100644 basic_info/migrations/0018_product_from_mdy.py create mode 100644 basic_info/migrations/0019_customer_mdy_fields.py diff --git a/api_v1/admin.py b/api_v1/admin.py index 2cca964..fca6214 100644 --- a/api_v1/admin.py +++ b/api_v1/admin.py @@ -1,6 +1,6 @@ from django.contrib import admin from django.contrib.admin import action -from api_v1.models import UploadedFile +from api_v1.models import UploadedFile, DataSync from .tasks import backup_database @@ -21,3 +21,13 @@ class UploadedFileAdmin(admin.ModelAdmin): @action(description='备份数据库') def backup_database(self, request, queryset): backup_database.delay() + + +@admin.register(DataSync) +class DataSyncAdmin(admin.ModelAdmin): + list_display = ['id', 'table_name', 'page_index', 'page_size', 'total_count', 'synced_rows', 'last_ctime', 'last_rowid', 'note'] + list_filter = ['table_name', 'created_at'] + search_fields = ['table_name', 'last_rowid', 'note'] + readonly_fields = ['created_at', 'updated_at'] + date_hierarchy = 'created_at' + ordering = ['-created_at'] diff --git a/api_v1/migrations/0003_datasync.py b/api_v1/migrations/0003_datasync.py new file mode 100644 index 0000000..9a44f69 --- /dev/null +++ b/api_v1/migrations/0003_datasync.py @@ -0,0 +1,33 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('api_v1', '0002_alter_uploadedfile_path'), + ] + + operations = [ + migrations.CreateModel( + name='DataSync', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')), + ('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')), + ('table_name', models.CharField(choices=[('product', '产品'), ('customer', '客户')], max_length=50, 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='远端总数')), + ('synced_rows', models.PositiveIntegerField(default=0, verbose_name='本次同步数量')), + ('last_ctime', models.DateTimeField(blank=True, null=True, verbose_name='最新数据时间')), + ('last_rowid', models.CharField(blank=True, max_length=64, verbose_name='最新 RowID')), + ('note', models.CharField(blank=True, max_length=255, verbose_name='备注')), + ], + options={ + 'verbose_name': '数据同步记录', + 'verbose_name_plural': '数据同步记录', + 'db_table': 'api_data_sync', + 'ordering': ['-created_at'], + }, + ), + ] diff --git a/api_v1/models.py b/api_v1/models.py index 4d2c1d3..83965aa 100644 --- a/api_v1/models.py +++ b/api_v1/models.py @@ -82,3 +82,29 @@ class UploadedFile(ModelBase): if self.path: return self.path.url return None + + +class DataSync(ModelBase): + """记录外部数据同步的执行情况""" + + class TableName(models.TextChoices): + PRODUCT = 'product', '产品' + CUSTOMER = 'customer', '客户' + + table_name = models.CharField(max_length=50, choices=TableName.choices, 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='远端总数') + synced_rows = models.PositiveIntegerField(default=0, verbose_name='本次同步数量') + last_ctime = models.DateTimeField(null=True, blank=True, verbose_name='最新数据时间') + last_rowid = models.CharField(max_length=64, blank=True, verbose_name='最新 RowID') + note = models.CharField(max_length=255, blank=True, verbose_name='备注') + + class Meta: + db_table = 'api_data_sync' + verbose_name = '数据同步记录' + verbose_name_plural = '数据同步记录' + ordering = ['-created_at'] + + def __str__(self): + return f'{self.table_name} @ {self.created_at:%Y-%m-%d %H:%M:%S}' diff --git a/api_v1/tasks.py b/api_v1/tasks.py index cee1f4f..6456c65 100644 --- a/api_v1/tasks.py +++ b/api_v1/tasks.py @@ -1,7 +1,11 @@ +import asyncio +import json import logging import os import shutil import subprocess +from datetime import datetime +from decimal import Decimal, InvalidOperation from pathlib import Path from celery import shared_task @@ -9,47 +13,16 @@ from django.conf import settings from django.utils import timezone from basic_info import models as basic_models +from api_v1 import models as api_models +from flower.utils import ( + fetch_products_from_mingdaoyun, + fetch_customers_from_mingdaoyun, +) logger = logging.getLogger(__name__) -@shared_task(bind=True) -def ping_task(self, message: str = 'ping'): - """ - 最简单的心跳任务，用于验证 Celery worker 是否能够 - 正确消费队列并返回结果。 - """ - payload = { - 'task_id': self.request.id, - 'message': message, - 'timestamp': timezone.now().isoformat(), - } - logger.info('Celery ping_task 执行成功: %s', payload) - return payload - - -@shared_task(bind=True) -def merchant_product_count(self, merchant_id: int): - """ - 计算指定商户下的产品数量，用于演示如何在任务中访问数据库。 - """ - count = basic_models.Product.objects.filter(merchant_id=merchant_id).count() - payload = { - 'task_id': self.request.id, - 'merchant_id': merchant_id, - 'product_count': count, - 'calculated_at': timezone.now().isoformat(), - } - logger.info( - 'Celery merchant_product_count 统计完成: merchant=%s count=%s task=%s', - merchant_id, - count, - self.request.id, - ) - return payload - - def _ensure_backup_dir(output_dir: str | None) -> Path: base_dir = Path(settings.BASE_DIR) backup_dir = Path(output_dir) if output_dir else (base_dir / 'data-bak') @@ -124,3 +97,288 @@ def backup_database(self, output_dir: str | None = None, filename_prefix: str = logger.info('数据库备份完成: %s', payload) return payload + +def _get_mdy_merchant(): + merchant_id = getattr(settings, 'MDY_MERCHANT_ID', None) + qs = basic_models.Merchant.objects.all() + if merchant_id: + qs = qs.filter(id=merchant_id) + merchant = qs.order_by('id').first() + if not merchant: + raise RuntimeError('未找到用于明道云同步的商户，请先创建商户或配置 MDY_MERCHANT_ID') + return merchant + + +def _get_mdy_category(merchant): + category_id = getattr(settings, 'MDY_PRODUCT_CATEGORY_ID', None) + qs = basic_models.ProductCategory.objects.filter(merchant=merchant) + if category_id: + qs = qs.filter(id=category_id) + category = qs.order_by('id').first() + if not category: + raise RuntimeError('未找到用于明道云同步的产品类别，请先创建类别或配置 MDY_PRODUCT_CATEGORY_ID') + return category + + +def _parse_mdy_datetime(value: str | None): + if not value: + return None + try: + dt = datetime.strptime(value, '%Y-%m-%d %H:%M:%S') + except ValueError: + return None + if timezone.is_naive(dt): + dt = timezone.make_aware(dt, timezone.get_current_timezone()) + return dt + + +def _map_unit(unit_label: str | None) -> int: + if not unit_label: + return basic_models.ProductUnitEnum.METER + mapping = { + '米': basic_models.ProductUnitEnum.METER, + '码': basic_models.ProductUnitEnum.YARD, + '公斤': basic_models.ProductUnitEnum.KG, + } + return mapping.get(unit_label, basic_models.ProductUnitEnum.METER) + + +def _ensure_decimal(value: str | None): + if not value: + return None + try: + return Decimal(str(value)) + except (InvalidOperation, TypeError, ValueError): + return None + + +def _run_fetch(page: int, page_size: int): + return asyncio.run(fetch_products_from_mingdaoyun(page=page, page_size=page_size)) + + +def _run_fetch_customers(page: int, page_size: int): + return asyncio.run(fetch_customers_from_mingdaoyun(page=page, page_size=page_size)) + + +def _upsert_product(product_data, merchant, category): + description = '' + if product_data.detail: + description = json.dumps(product_data.detail, ensure_ascii=False) + defaults = { + 'category': category, + 'name': product_data.name or product_data.uid, + 'color': product_data.color or '', + 'description': description, + 'unit': _map_unit(product_data.unit), + 'from_mdy': True, + } + width_decimal = _ensure_decimal(product_data.width) + if width_decimal is not None: + defaults['width_size'] = width_decimal + + product_obj, created = basic_models.Product.objects.get_or_create( + merchant=merchant, + human_id=product_data.uid, + defaults=defaults, + ) + if created: + return True + + if not product_obj.from_mdy: + return False + + updated = False + for field, value in defaults.items(): + if getattr(product_obj, field) != value: + setattr(product_obj, field, value) + updated = True + if updated: + product_obj.save(update_fields=list(defaults.keys())) + return updated + + +@shared_task(bind=True) +def sync_mdy_products(self, page_size: int = 300, max_pages: int = 5, max_records: int | None = None): + """ + 从明道云同步产品数据 + """ + merchant = _get_mdy_merchant() + category = _get_mdy_category(merchant) + max_records = max_records or page_size + + last_sync = api_models.DataSync.objects.filter( + table_name=api_models.DataSync.TableName.PRODUCT + ).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 '' + + synced_rows = 0 + page_index = 1 + total_count = 0 + latest_ctime = last_ctime + latest_rowid = last_rowid + reached_existing = False + + while page_index <= max_pages and synced_rows < max_records: + products, total = _run_fetch(page_index, page_size) + total_count = total + if not products: + break + + for item in products: + product_ctime = _parse_mdy_datetime(item.created_at) + if last_ctime and product_ctime: + if product_ctime < last_ctime: + reached_existing = True + break + if product_ctime == last_ctime and last_rowid and item.rowid == last_rowid: + reached_existing = True + break + + 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 synced_rows >= max_records: + break + + if reached_existing or synced_rows >= max_records or 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, + 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='reached existing data' if reached_existing else '', + ) + + payload = { + 'task_id': self.request.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('明道云产品同步完成: %s', payload) + return payload + + +def _upsert_customer(customer_data, merchant): + mdy_uid = customer_data.uid or customer_data.rowid + if not mdy_uid: + return False + defaults = { + 'merchant': merchant, + 'name': customer_data.name or customer_data.uid, + 'area': customer_data.area or '', + 'from_mdy': True, + } + customer_obj, created = basic_models.Customer.objects.get_or_create( + mdy_uid=mdy_uid, + defaults=defaults, + ) + if created: + return True + + if not customer_obj.from_mdy: + return False + + updated_fields = {} + for field, value in defaults.items(): + if getattr(customer_obj, field) != value: + updated_fields[field] = value + + if updated_fields: + for field, value in updated_fields.items(): + setattr(customer_obj, field, value) + customer_obj.save(update_fields=list(updated_fields.keys())) + return True + return False + + +@shared_task(bind=True) +def sync_mdy_customers(self, page_size: int = 300, max_pages: int = 5, max_records: int | None = None): + """ + 从明道云同步客户数据 + """ + merchant = _get_mdy_merchant() + max_records = max_records or page_size + + last_sync = api_models.DataSync.objects.filter( + table_name=api_models.DataSync.TableName.CUSTOMER + ).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 '' + + synced_rows = 0 + page_index = 1 + total_count = 0 + latest_ctime = last_ctime + latest_rowid = last_rowid + reached_existing = False + + while page_index <= max_pages and synced_rows < max_records: + customers, total = _run_fetch_customers(page_index, page_size) + total_count = total + if not customers: + break + + for item in customers: + record_ctime = _parse_mdy_datetime(item.created_at) + if last_ctime and record_ctime: + if record_ctime < last_ctime: + reached_existing = True + break + if record_ctime == last_ctime and last_rowid and item.rowid == last_rowid: + reached_existing = True + break + + 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 synced_rows >= max_records: + break + + if reached_existing or synced_rows >= max_records or 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, + 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='reached existing data' if reached_existing else '', + ) + + payload = { + 'task_id': self.request.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('明道云客户同步完成: %s', payload) + return payload + diff --git a/api_v1/views/printing/views.py b/api_v1/views/printing/views.py index fece15f..4cf46b9 100644 --- a/api_v1/views/printing/views.py +++ b/api_v1/views/printing/views.py @@ -516,6 +516,7 @@ class PlateOrderFilterSet(django_filters.FilterSet): """开版订单过滤器""" customer_name = django_filters.CharFilter(field_name='customer__name', lookup_expr='icontains') customer_phone = django_filters.CharFilter(field_name='customer__mobile', lookup_expr='icontains') + image_name = django_filters.CharFilter(lookup_expr='icontains') salesperson = django_filters.NumberFilter() merchandiser = django_filters.NumberFilter() plate_type = django_filters.CharFilter(lookup_expr='icontains') @@ -546,6 +547,7 @@ class PlateOrderViewSet(viewsets.ModelViewSet): list: 获取开版订单列表 retrieve: 获取开版订单详情 create: 创建开版订单 + image_name: 图片名称（模糊查询） update: 更新开版订单 partial_update: 部分更新开版订单 invalidate: 作废开版订单 diff --git a/basic_info/admin.py b/basic_info/admin.py index 90b549a..a7b1661 100644 --- a/basic_info/admin.py +++ b/basic_info/admin.py @@ -60,7 +60,7 @@ class UserProfileAdmin(AdminBase): @admin.register(models.Merchant) class MerchantAdmin(admin.ModelAdmin): - list_display = ('name', 'type', 'email', 'mobile', 'auto_complete_stock_change') + list_display = ('id', 'name', 'type', 'email', 'mobile', 'auto_complete_stock_change') search_fields = ('name', 'mobile') list_filter = ('type', 'auto_complete_stock_change') diff --git a/basic_info/migrations/0018_product_from_mdy.py b/basic_info/migrations/0018_product_from_mdy.py new file mode 100644 index 0000000..6c2bf12 --- /dev/null +++ b/basic_info/migrations/0018_product_from_mdy.py @@ -0,0 +1,20 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('basic_info', '0017_alter_customer_mobile'), + ] + + operations = [ + migrations.AddField( + model_name='product', + name='from_mdy', + field=models.BooleanField( + default=False, + help_text='标记该产品是否通过明道云接口同步', + verbose_name='是否来自明道云同步', + ), + ), + ] diff --git a/basic_info/migrations/0019_customer_mdy_fields.py b/basic_info/migrations/0019_customer_mdy_fields.py new file mode 100644 index 0000000..6f4d693 --- /dev/null +++ b/basic_info/migrations/0019_customer_mdy_fields.py @@ -0,0 +1,21 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('basic_info', '0018_product_from_mdy'), + ] + + operations = [ + migrations.AddField( + model_name='customer', + name='mdy_uid', + field=models.CharField(blank=True, help_text='对应明道云记录的唯一标识', max_length=100, null=True, unique=True, verbose_name='明道云UID'), + ), + migrations.AddField( + model_name='customer', + name='from_mdy', + field=models.BooleanField(default=False, help_text='标记该客户是否通过明道云接口同步', verbose_name='是否来自明道云同步'), + ), + ] diff --git a/basic_info/models.py b/basic_info/models.py index bac85af..68e1388 100644 --- a/basic_info/models.py +++ b/basic_info/models.py @@ -261,6 +261,11 @@ class Product(ModelBase): null=True, verbose_name='公斤转换率', ) + from_mdy = models.BooleanField( + default=False, + verbose_name='是否来自明道云同步', + help_text='标记该产品是否通过明道云接口同步', + ) def __str__(self): return self.name @@ -318,6 +323,19 @@ class Customer(ModelBase): email = models.EmailField(blank=True, null=True, verbose_name='电子邮箱') contact = models.CharField(blank=True, null=True, verbose_name='联系人') description = models.TextField(blank=True, null=True, verbose_name='备注描述') + mdy_uid = models.CharField( + max_length=100, + unique=True, + null=True, + blank=True, + verbose_name='明道云UID', + help_text='对应明道云记录的唯一标识', + ) + from_mdy = models.BooleanField( + default=False, + verbose_name='是否来自明道云同步', + help_text='标记该客户是否通过明道云接口同步', + ) def __str__(self): return self.name diff --git a/celerybeat-schedule b/celerybeat-schedule index e8e9560be005761a6d41b52c34ed247ebed84a5b..bc56a4619d866683a9a18cfedbd862971421f222 100644 GIT binary patch delta 514 zcmZojXh_(=D!|Qin}Poh|8f3BJhwL%p5~Emb#63fk?N{R%_}L&Of7C;ow}BZ0Ssy< z=gLV>_Luw4v(?ab;felnfRPETTPf zi3OSQWrlhsiN)E)dLWhYNLq`hWNN&Dtp%J)HFgiRr2F#hF#9Q@m4pICB##;z1(CQ@ojaxIw(4)a3l4 z6d?aZY3&q{)*eBSt;wauCHc9jMFHSx{jozqtUj5)+aJhK=mZ0!$@U3>$curI?`c$gmltDn7m} OwWv5VKQBI>VG{uNshaI%P;uSBt)kN~riNNGuOd}>}|Qch}018aw#BLC$3 z@`_AKf|HlZD+zD`O=96c%fSDE|1AGY{+pWx6?X7X3}7|pW|m@vS}DxWti*_9iU1q4 X0AooNKS($}zAUw9uXHtMFkgJad#BK1rY^T+;ANiT)}w25+yYYa*k3~5Ry3CSi}Vj)z)vIoUB7d;HKyOYeYug=Vd zY%n$v6sdy;I}r~pLhZqeV5!>HmkPbqmupo_d+MbKiXe!15ue%Yrkju$S9(ZxzXJ=i z&&)IPe?FNV$nayH@ly}HDjmMFUy?$Sw}iu!UtasRZ|d#-$0F~I-FR0tHB`38-hW5E zdMe%1E*cx#Zb`x0!5eM2g6F+`SP(z}0R#|0009ILKmdVVCh*#^rIE&-(=Tfo+cY$5 z(ZBp*@n0ZGT5$WF2NSB1DkurnR%5EADKRyk%;#LEHLj&JvmnPb)pnis6LDQjt|Ob(+PQG}}mPm8)mmV=YZTw>9rQC~ODBmRR0M z*+#~4ou(PhR&Co9=X9Fqvl-3EOlQUF3u5)Sw<_8LNncezV1FRs-3h9b?}6) zW@dENaHD&?R__z_Iib?HtkQtIQ^d_|MhxE@R#X!E>=|(NUcJ{AJr>;9L64~=o40d$ zTS*wEIR2oT%ME&Dhhu8eb%N7IO3SEeP05+sv@s_tO2ovbqGW!-^$z5O(usoDo=xQA zwqg}BaaV8b-=x)fWF4Tlr&7_?a`nI}R(F>q`Ak>uE{<$Oe^?an_yz`ti^Z=mKaf*r zG{rLJH8+}Y0%>(l@mef5+TgT$-%Kr@H520d*X4RUypzWBmYq#&=DmV9X?7mnyy9}_ z^(9M3%O!`3CF`OGFm!KJQb0Uf}m%>PwC< zJ1wHIG5C!X{4IDVxFS|q5I_I{1Q0*~0R#|0009ILKw#$zgnZ$Kisu5-#U@{Ywd4W353q-2U3!M9I-;ZB(ugs7a*z&%@1_Tg5009ILKmY**5I_I{1a_Xl zR(XL>s>=(k%>8v`GH3r$&I@?&7VzFNu=55?MF=2(00IagfB*srAb`NO0^|kA3)D|u zU~IGZ3sj#MIPli>5Ft7!o-F#!t#2q1s}0tg_000Iagu=@l=USOR02$+wcKIS91UwMJ# z&Dg{t>nM4F-FL&IVgwLC009ILKmY**5I_KdIu`g(d4bE@&PQEyG)y+8eV_g-GWOA9a`LH%7&j2Hm~ z5I_I{1Q0*~ftnB?FF;EzL(zWnREKa-sG6S0-L6=LJ^N0>Sh3!pJxgKmY**5I_I{1Q0-= zrUb|fkQb>m>9A(#LF literal 444992 zcmeI*d5~1~fxvO_fs7~&9>B;U3LY@Z0E&2kK~xaML=3BCEUmF;XrN&jaHdBEMOmMC zWvHy75#4wsYq@JoG|@yiR@`_6*F}wSRn&lM;=y|Ey6bJ!Y`38OlGUiMx+5>|gGKr5 znfdka_xZiwKjXLh?RWN;=X4qQbX}J&eY%wX^lv$?Y5Ig;T)y8y3+li5OgNy~ZJx8<_!OS`W;rR&D^FDZ2PoteqCENYyY$!D4}?b*hrO!K^sg@w+oo3kz1wnYt1 z*-XCB*=t7gob1ewmh9x>7>f!k3d2A32M1*q<{D2O(~!@!&ueekINZm>7dmT-@088w z=4YdqD+-79?D4`jHN8g^=T%?)N>fL!C7)|;FLd^poy}+R`L^PFcGfIj(3;J)&RS5s zJfV1bMCr2l`c{n{tqXI_^IEcn^9o-t&UR9JE;DMzoJ{NNIhkBxTG!I6rN7gH&OV#g z)xK##rMGBqThLlu{)fv7>PL+!bat&T%o$wzCtv)jC55IDONuM*xu7Gzup{3%GuKvp zk5QS03rCgMMmA;U6*_BYnfckqg>BhcxzmfYn%PwR(ubNIOA4id+8eg(6Aay< zBj4Ou{MM$H?98RzFKF1MHt2I$esOVIw%v4B{^Ii&xFVOe1dV$+-nE252<S@Mn5~ej7hu;4^!HZ8zNCenMySO&fcGAze1s0=-Mm zE%1i`0RjXF5FkK+009C7(kBoG{ksiVc4qmR1oJ(U;8UJSFs)1Rxdo*s5)7I%sM60P z_{;SI-#_=HXWlyh%B6aN^c{99CP07y0RjXF5FkK+0D*)HM0$b#fATzn&)W;s+%fi` zWk-)673l@8={>~r2oip0m_7jl1PBlyK!5-N0tDg;=mkFaXAJzMdx5&DJda@Sf8Fz6 zrp?`ATfIQsBje`?5FkK+009C72oNAZfWVg|@Y%gU-Df?IpvPg|R!w^B%~#63K&ck+ zJc2K2N4+fp0t5&UAV7cs0RjZRYyrK%X3-1m_zBM=s7^2NwVn6*+8=&3_P_N4U-r@S z{sagRAV7cs0RjXF5FkL{3kv+1USP+K&m;KEUSREx6YhI^>AyeyQ7^Eu7N}kS1!v;h z2@oJafB*pk1PBlyK;W-KKrfKwUSL2lbZleesoA#n+=AA|#--g?Znf#Q?Cid8!?=dg zg@)pP2c^ILTaIg*KH(RaKQjMkbI%{L@?O2bU&jHq3IYTO5FkK+009C72oOk-z!&cY z1{8I+C<%S^dx7x>y;n009C72oNAZfB=EAKo|_{ zHelJA+17kpF56!GAI|f;Zv5hsLTBHZnOw`F#+jLXrYX~&ZEVUk&+AxN=-j$F+mdZt z)XW;D;q&g^K(PA-nIsIa0i{8N8$P-bDS@zgO5`Aqw~_J)naeLQ@jv!?h? z*?ew(HhQ_DaA?mSFKkoOdqiv#|xz_eVXOG$0d?uf7E52uE&Ef^E*<9i-jCRdo&we)J~@3f$^ z&!%;?Z(304Et=aFv=*0NT2?imM-cP#1!D99S6*N9^K-|aw^%PwJ_N1^5FkK+009C7 z2oNAZfIuPyBE7)C(t`&&^39Fe)=X1NcIKyizQF(M1=h44*>S|&{DFFbMC`ap5+Fc; z009C72oNAZpxOdqFi0d{VK7)PkfvT>{_73j8J)jk znqDB82iB|!5FkK+009C72oNAZAjJZaUSM#{Uf^f{cEI&h9(wwAy+DfZBTFYhfB*pk z1PBlyK!8A5APjcW3#6$R81vKM#T84od{-|}J_N1^5FkK+009C72oNAZfIuPyBE7&) zF?)d{4|)0C-@du-X}v%qcHATh5FkK+009C72oNApZGkWtq8CV0FEIUuvxY6-uIY5W zK(&vS@d*$hK!5-N0t5&UAV7dXg#{wLz>t`|z}#Q%wY>i3fqnD>72X@e5gfi-Ia1PBlyK!5-N0t5&UNU=bq7uY#w zFL3xN+n@MYmpixB3#9lyvUCCj2oNAZfB*pk1PGJ`!eAG@K$?1i$Bw;WM%UVBC+P*s zhrks90t5&UAV7cs0RjXF5J-eTq!-vFW-oBn(sTQ^PPkycULX-WZjuBD5FkK+009C7 z2oR{YKo|_w3#6$R$bEHq&!^scWRPB<+DFUy1PBlyK!5-N0t5&UAV8qP0+C)|Xv|*V zp7*z?dw1E87QH})_r`Dp2oNAZfB*pk1PBmFra%}B(+i}j7r5t!5B9ur)#V%X0?9nE zW=((q0RjXF5FkK+009Ci7Kro$!(#RVgS(EHJZkHk`soEyd>>gl0RjXF5FkK+009C7 z$^v1qt6m^Yy+F-5KUw>~H?|+H7bqVBR|E(UAV7cs0RjXF5FkJx5dx83VAq(vz|==q zzjw&ByY8bGNW_ktBmn{h2oNAZfB*pk1gb3%2D|A6($ovQvDeriG=CmUsGD=`72T#bAEg&a#EzRJ0RjXF5FkK+009C7sx1%( z!}S7b>IEjQ>AT^yv!CBqFHr5HWqbkz2oNAZfB*pk1PBlyP+@^cFEBi2FL2+m5fc{7 z*`m8%pu&4&I06I+5FkK+009C72qaS=3`Xb$($ovQ{_c)f?|IKr&3%vN~2Lpfe)EnFB1*(0tj8A|70RjXF z5FkK+009C7Dl8D`1@?^D3w-64Z?8P^sk)o=0u|mH!x11rfB*pk1PBlyKp>d{VX&88 zAWgkM_TC>}a#{ag_vr_(R*$X^0Ve+86N4|c( zULeKyk);zLK!5-N0t5&UAV8oj5C$Xl0%__6`p(^V{FCh?@74>H4}mKJ1PBlyK!5-N z0t5&UAdm=wNG~ulW-svGeZTM3<8MZ<(hDSF$4!y|0RjXF5FkK+009Ek76^j|y+E3J zf&Gqny#B3qj}6reRQqTdp8x>@1PBlyK!5-N0t5(DSRm31G{o!$j-1x(-oGEdL%m+0 z!h2&l0t5&UAV7cs0RjXFBvT*^M(G98)C>IAvWDQ$Nyq(KFObXwYt{q^5FkK+009C7 z2oNBUVu466Fe+v*Fl^${BM;7=IaM!^;`_+b2@oJafB*pk1PBlyP!|gSclqF9$KI+JNW_kt zBmn{h2oNAZfB*pk1gb3%24nOBY3c>89Q9QD#O?oMkzSzMN6YvG2oNAZfB*pk1PBly zK%l|`kzQa-%wFK?HKX46`7?)i*9%m5ZwyC(009C72oNAZfB=DH3WUMldVw_c0>}Ps z?=|_xsonGf$vm)TO@IIa0t5&UAV7cs0Rkx&i1Y$`$Ls}me1Gz#y*hj5^#Uork1U-4 z0RjXF5FkK+009DJfiT!dFOa5Q;Kz0MeslauSAL)uC?5h>1PBlyK!5-N0t5&UAV44y z0+C)|pP0SC{k8XGpBwOYZ@oYwcHATh5FkK+009C72oNApZGkWts~1R9FEIS5$JQRW z)$6r-fodNu;}al2fB*pk1PBlyK!5;&3JXMffw3`rfy+lvx$T@cCVodRP~p8X9039Z z2oNAZfB*pk1d=Hb2K(v-($otav*Fcg+tuxKm0lp32iB|!5FkK+009C72oNAZAjJZa zUSQvty}-o}-MysylCNK_7fA7aWa$J55FkK+009C72oNX>guytyK$?1icly2A^R(w$ z*Xjkzhrks90t5&UAV7cs0RjXF5J-eTq!$<$vlkfo{JS+bzx+X~ULX-WZjuBD5FkK+ z009C72oR{YKp5<&7f4etaAW?tpFF<)!OQgm)jnFrCqRGz0RjXF5FkK+009CO7Kro$ z`^D@9uJ3j7K@%3<(oHW=;k_{&0RjXF5FkK+009C7k|_`d`|AbL)C+w7*%N2gb$k3W zy+AS#tXUHvK!5-N0t5&UAV7dXiUlIQ!2U6Nf%fpWGsg_-9H|#b@qJ|J1PBlyK!5-N z0t5&UC<}za0eXQn^#V`4cKh997Ej+-FHk-Nt_TnyK!5-N0t5&UAV7dXA_OA6zyUFP zfy=s_bj=UGbcAWgl%JC}8T@tD)n|0t5&UAV7cs z0RqVs2!ru@fi(33j5Ui{s9y+DfZBTFYhfB*pk1PBlyK!8A5APgqx1=7?DoYL=&OP)CHmhJTdoy#z7f9%dHEjX} z2oNAZfB*pk1PBmFuRx?1_)5%P;OJLRzjxKH=ePNMf%M)+R!)Eb0RjXF5FkK+0D*`= z7);a)q^TEp_n7maY0cg=TrUtg1uh8?AV7cs0RjXF5FkK+KoSHZy}-nny}+`a2R?A} zizhGD3nXE~&5-~B0t5&UAV7cs0RmMQ2!n(50%__6esk8kSr4^*utYCV^|R#-2oNAZ zfB*pk1PBlyK!8AH1tPt`!7+P*Jx{%L{nB?jGJ1i^ZjI3h5FkK+009C72oNBUP=PR* zq!&n2FYwZu#VfYjIM40@P~`e8Z+Q!y+Gs?xFkS;009C7 z2oNAZfB*pkNf3zi0*A!x1>U%7MqTELU!SQLNWzAjBLM;g2oNAZfB*pk1gb6&28Zee z($ovA%HQ_VyjzF&(+gDnYNH1_`%wFKa6P7OvZ_I3| z7pUyk7>xh{0t5&UAV7cs0RjmX2!q4)0%__6mXBM#cF3d8Jf;^&=!rFL0t5&UAV7cs z0RjXF5J;~;q!&0WW-oBYF%O+tKkU3fFOc5b$jS*2AV7cs0RjXF5Fii{2!qLbfi(33 zckZ$BrE9x?Zjjdq z;pRwy009C72oNAZfB=E23xvTGy+E3JfhkkIapXn+@W5GmfvTS^Z$N+m0RjXF5FkK+ z009C7Dk~7_1*XL81(v!X&K{>1sO;7ljQ{}x1PBlyK!5-N0tpoegTwU#Y3c>` z$v(2;z=wP1^#Tb!v8GLc009C72oNAZfB*pk=@p3d0*A-!1zL`}^T|V6@4HPeklx$K z$_Wr4K!5-N0t5&UAP^AfxkOeFOY-{H%9^l2oNAZfB*pk1PD}JAPkPs3#6$R_|Y8? pK9d{Q@`_%d>SxOv5FkK+009C72oNAZfB=EY3PgH=BR*#@@PE$I5+DEo diff --git a/flower/settings.py b/flower/settings.py index cd62a79..95d3449 100644 --- a/flower/settings.py +++ b/flower/settings.py @@ -53,6 +53,13 @@ CSRF_TRUSTED_ORIGINS = env.list('CSRF_TRUSTED_ORIGINS', default=[ 'https://yuwenerp.yuwen.cloud', ]) +# 明道云同步配置 +MDY_MERCHANT_ID = env.int('MDY_MERCHANT_ID', default=1) +MDY_PRODUCT_CATEGORY_ID = env.int('MDY_PRODUCT_CATEGORY_ID', default=1) +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) + # CORS 配置 CORS_ALLOW_ALL_ORIGINS = DEBUG # 开发环境允许所有源，生产环境需要配置白名单 @@ -303,4 +310,22 @@ CELERY_BEAT_SCHEDULE = { 'filename_prefix': 'db-backup', }, }, + 'mdy_product_sync': { + 'task': 'api_v1.tasks.sync_mdy_products', + 'schedule': crontab(hour='*/1', minute=0), + 'kwargs': { + 'page_size': MDY_SYNC_PAGE_SIZE, + 'max_pages': MDY_SYNC_MAX_PAGES, + 'max_records': MDY_SYNC_MAX_RECORDS, + }, + }, + 'mdy_customer_sync': { + 'task': 'api_v1.tasks.sync_mdy_customers', + 'schedule': crontab(hour='*/1', minute=0), + 'kwargs': { + 'page_size': MDY_SYNC_PAGE_SIZE, + 'max_pages': MDY_SYNC_MAX_PAGES, + 'max_records': MDY_SYNC_MAX_RECORDS, + }, + }, } diff --git a/flower/utils.py b/flower/utils.py index 28a1cf4..6611bb3 100644 --- a/flower/utils.py +++ b/flower/utils.py @@ -1,227 +1,283 @@ -# import aiohttp -# from typing import Optional, Dict, Any, List -# from pydantic import BaseModel, Field, computed_field -# from enum import Enum -# import json +import aiohttp +from typing import Optional, Dict, Any, List +from pydantic import BaseModel, Field, computed_field +from enum import Enum +import json -# class Product(BaseModel): -# """产品模型 - 请在此填入你的字段""" -# uid: str -# name: str -# unit: str | None -# color: str | None -# detail_str: str | None = Field(exclude=True) # 在序列化时排除此字段 -# created_at: str +class Product(BaseModel): + """产品模型 - 请在此填入你的字段""" + uid: str + rowid: str + name: str + unit: str | None + color: str | None + width: str | None = None + detail_str: str | None = Field(exclude=True) # 在序列化时排除此字段 + created_at: str -# @computed_field -# @property -# def detail(self) -> dict[str, Any]: -# """将 detail_str 转换为字典""" -# try: -# return json.loads(self.detail_str) if self.detail_str else {} -# except (json.JSONDecodeError, TypeError): -# return {} + @computed_field + @property + def detail(self) -> dict[str, Any]: + """将 detail_str 转换为字典""" + try: + return json.loads(self.detail_str) if self.detail_str else {} + except (json.JSONDecodeError, TypeError): + return {} -# class ProductListResponse(BaseModel): -# """产品列表响应""" -# products: List[Product] -# total: int +class ProductListResponse(BaseModel): + """产品列表响应""" + products: List[Product] + total: int -# mdy_table_map = { -# 'product': 'spmx', -# 'customer': 'quanbu', +mdy_table_map = { + 'product': 'spmx', + 'customer': 'quanbu', +} + + +class HTTPMethod(Enum): + """HTTP 请求方法枚举""" + GET = "GET" + POST = "POST" + + +class MingDaoYunClient: + """明道云 API 客户端""" + + def __init__(self, app_key: str, sign: str, base_url: str = ""): + """ + 初始化客户端 + + Args: + app_key: 应用密钥 + sign: 签名 + base_url: API 基础地址 + """ + self.app_key = app_key + self.sign = sign + self.base_url = base_url + + async def request( + self, + method: HTTPMethod, + endpoint: str, + data: Optional[Dict[str, Any]] = None, + params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Any: + """ + 通用异步请求函数 + + Args: + method: 请求方法 (GET/POST) + endpoint: API 端点路径 + data: POST 请求体数据（JSON） + params: URL 查询参数 + headers: 自定义请求头 + + Returns: + 响应 JSON 数据 + + Raises: + aiohttp.ClientError: 请求失败时抛出 + """ + url = f"{self.base_url}{endpoint}" + + # 构建默认请求头 + default_headers = { + "Content-Type": "application/json", + } + + if headers: + default_headers.update(headers) + + # 添加认证参数 + auth_params = { + "appKey": self.app_key, + "sign": self.sign + } + + async with aiohttp.ClientSession() as session: + if method == HTTPMethod.GET: + async with session.get(url, params=params, json=auth_params) as response: + response.raise_for_status() + return await response.json() + + elif method == HTTPMethod.POST: + if data is None: + data = {} + + data.update(auth_params) + async with session.post(url, json=data) as response: + response.raise_for_status() + return await response.json() + + else: + raise ValueError(f"Unsupported HTTP method: {method}") + + async def get( + self, + endpoint: str, + params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Any: + """ + 发送 GET 请求 + + Args: + endpoint: API 端点路径 + params: URL 查询参数 + headers: 自定义请求头 + + Returns: + 响应 JSON 数据 + """ + return await self.request( + method=HTTPMethod.GET, + endpoint=endpoint, + params=params, + headers=headers, + ) + + async def post( + self, + endpoint: str, + data: Optional[Dict[str, Any]] = None, + params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Any: + """ + 发送 POST 请求 + + Args: + endpoint: API 端点路径 + data: POST 请求体数据（JSON） + params: URL 查询参数 + headers: 自定义请求头 + + Returns: + 响应 JSON 数据 + """ + return await self.request( + method=HTTPMethod.POST, + endpoint=endpoint, + data=data, + params=params, + headers=headers, + ) + + +product_type_map = { + 'name': '668caa5eb80969563ecaee7d', + 'color': '66ed3b9ae01d5599bdb45f6d', + 'uid': '668caa5eb80969563ecaee7c', + 'unit': '668e2ee2ca758c8cc5d0dc1d', + 'width': '668e2194790c0e04058b4f3c', + 'detail_str': '668caa5eb80969563ecaee7e', + 'created_at': 'ctime', + 'rowid': 'rowid', +} + +customer_type_map = { + 'name': '62d52f4b8d2972284492dcf9', + 'area': '62d52f4b8d2972284492dd09', + 'created_at': 'ctime', + 'rowid': 'rowid', + 'uid': '668bb9370207cf7520fe551e', +} + +class Customer(BaseModel): + """客户模型 - 请在此填入你的字段""" + uid: str + rowid: str + name: str + area: str | None + created_at: str + + +def pick_customer(fields: dict) -> Customer: + """从字段字典中提取客户信息""" + data = {k: fields.get(v, '') for k, v in customer_type_map.items()} + return Customer(**data) + + +# json param example: +# { +# "appKey": "208e55fea5cea59f", +# "sign": "MWU0YmViYjkwZmM1ZDIzYzRiN2U3ZGQ4MmE4ZGNkMjc0MWM1ZmQ2ZjkwMjljODE4YmNkZTBhMzA0OTU2YzE2NA==", +# "worksheetId": "quanbu", +# "listType": 1, +# "sortId": "ctime", +# "isAsc": false, +# "notGetTotal": true # } -# class HTTPMethod(Enum): -# """HTTP 请求方法枚举""" -# GET = "GET" -# POST = "POST" +def pick_product(fields: dict) -> Product: + """ + 从字段字典中提取产品信息 + """ + data = {k: fields.get(v, '') for k, v in product_type_map.items()} + return Product(**data) -# class MingDaoYunClient: -# """明道云 API 客户端""" +async def fetch_products_from_mingdaoyun(page: int = 1, page_size: int = 100) -> tuple[list[Product], int]: + """ + 从明道云获取产品列表 -# def __init__(self, app_key: str, sign: str, base_url: str = ""): -# """ -# 初始化客户端 + Returns: + tuple: (产品列表, 总数量) + """ + client = MingDaoYunClient( + app_key='208e55fea5cea59f', + sign='MWU0YmViYjkwZmM1ZDIzYzRiN2U3ZGQ4MmE4ZGNkMjc0MWM1ZmQ2ZjkwMjljODE4YmNkZTBhMzA0OTU2YzE2NA==', + base_url="https://api.mingdao.com" + ) + response = await client.post( + endpoint='/v2/open/worksheet/getFilterRows', + data={ + 'worksheetId': 'spmx', + 'pageIndex': page, + 'pageSize': page_size, + 'sortId': 'ctime', + 'isAsc': False, + } + ) + data = response.get('data') + if not data: + return [], 0 -# Args: -# app_key: 应用密钥 -# sign: 签名 -# base_url: API 基础地址 -# """ -# self.app_key = app_key -# self.sign = sign -# self.base_url = base_url - -# async def request( -# self, -# method: HTTPMethod, -# endpoint: str, -# data: Optional[Dict[str, Any]] = None, -# params: Optional[Dict[str, Any]] = None, -# headers: Optional[Dict[str, str]] = None, -# ) -> Any: -# """ -# 通用异步请求函数 - -# Args: -# method: 请求方法 (GET/POST) -# endpoint: API 端点路径 -# data: POST 请求体数据（JSON） -# params: URL 查询参数 -# headers: 自定义请求头 - -# Returns: -# 响应 JSON 数据 - -# Raises: -# aiohttp.ClientError: 请求失败时抛出 -# """ -# url = f"{self.base_url}{endpoint}" - -# # 构建默认请求头 -# default_headers = { -# "Content-Type": "application/json", -# } - -# if headers: -# default_headers.update(headers) - -# # 添加认证参数 -# auth_params = { -# "appKey": self.app_key, -# "sign": self.sign -# } - -# async with aiohttp.ClientSession() as session: -# if method == HTTPMethod.GET: -# async with session.get(url, params=params, json=auth_params) as response: -# response.raise_for_status() -# return await response.json() - -# elif method == HTTPMethod.POST: -# if data is None: -# data = {} - -# data.update(auth_params) -# async with session.post(url, json=data) as response: -# response.raise_for_status() -# return await response.json() - -# else: -# raise ValueError(f"Unsupported HTTP method: {method}") - -# async def get( -# self, -# endpoint: str, -# params: Optional[Dict[str, Any]] = None, -# headers: Optional[Dict[str, str]] = None, -# ) -> Any: -# """ -# 发送 GET 请求 - -# Args: -# endpoint: API 端点路径 -# params: URL 查询参数 -# headers: 自定义请求头 - -# Returns: -# 响应 JSON 数据 -# """ -# return await self.request( -# method=HTTPMethod.GET, -# endpoint=endpoint, -# params=params, -# headers=headers, -# ) - -# async def post( -# self, -# endpoint: str, -# data: Optional[Dict[str, Any]] = None, -# params: Optional[Dict[str, Any]] = None, -# headers: Optional[Dict[str, str]] = None, -# ) -> Any: -# """ -# 发送 POST 请求 - -# Args: -# endpoint: API 端点路径 -# data: POST 请求体数据（JSON） -# params: URL 查询参数 -# headers: 自定义请求头 - -# Returns: -# 响应 JSON 数据 -# """ -# return await self.request( -# method=HTTPMethod.POST, -# endpoint=endpoint, -# data=data, -# params=params, -# headers=headers, -# ) + products = [pick_product(item) for item in data.get('rows', [])] + total_count = data.get('total', 0) + return products, total_count -# product_type_map = { -# 'name': '668caa5eb80969563ecaee7d', -# 'color': '66ed3b9ae01d5599bdb45f6d', -# 'uid': '668caa5eb80969563ecaee7c', -# 'unit': '668e2ee2ca758c8cc5d0dc1d', -# 'detail_str': '668caa5eb80969563ecaee7e', -# 'created_at': 'ctime', -# } +async def fetch_customers_from_mingdaoyun(page: int = 1, page_size: int = 100) -> tuple[list[Customer], int]: + """ + 从明道云获取客户列表 + """ + client = MingDaoYunClient( + app_key='208e55fea5cea59f', + sign='MWU0YmViYjkwZmM1ZDIzYzRiN2U3ZGQ4MmE4ZGNkMjc0MWM1ZmQ2ZjkwMjljODE4YmNkZTBhMzA0OTU2YzE2NA==', + base_url="https://api.mingdao.com" + ) + response = await client.post( + endpoint='/v2/open/worksheet/getFilterRows', + data={ + 'worksheetId': 'quanbu', + 'pageIndex': page, + 'pageSize': page_size, + 'sortId': 'ctime', + 'isAsc': False, + } + ) + data = response.get('data') + if not data: + return [], 0 - -# # json param example: -# # { -# # "appKey": "208e55fea5cea59f", -# # "sign": "MWU0YmViYjkwZmM1ZDIzYzRiN2U3ZGQ4MmE4ZGNkMjc0MWM1ZmQ2ZjkwMjljODE4YmNkZTBhMzA0OTU2YzE2NA==", -# # "worksheetId": "quanbu", -# # "listType": 1, -# # "sortId": "ctime", -# # "isAsc": false, -# # "notGetTotal": true -# # } - - -# def pick_product(fields: dict) -> Product: -# """ -# 从字段字典中提取产品信息 -# """ -# data = {k: fields.get(v, '') for k, v in product_type_map.items()} -# return Product(**data) - - -# async def fetch_products_from_mingdaoyun(page: int = 1, page_size: int = 100) -> tuple[list[Product], int]: -# """ -# 从明道云获取产品列表 - -# Returns: -# tuple: (产品列表, 总数量) -# """ -# client = MingDaoYunClient( -# app_key='208e55fea5cea59f', -# sign='MWU0YmViYjkwZmM1ZDIzYzRiN2U3ZGQ4MmE4ZGNkMjc0MWM1ZmQ2ZjkwMjljODE4YmNkZTBhMzA0OTU2YzE2NA==', -# base_url="https://api.mingdao.com" -# ) -# response = await client.post( -# endpoint='/v2/open/worksheet/getFilterRows', -# data={ -# 'worksheetId': 'spmx', -# 'pageIndex': page, -# 'pageSize': page_size, -# } -# ) -# data = response.get('data') -# if not data: -# return [], 0 - -# products = [pick_product(item) for item in data.get('rows', [])] -# total_count = data.get('total', 0) -# return products, total_count + customers = [pick_customer(item) for item in data.get('rows', [])] + total_count = data.get('total', 0) + return customers, total_count