forked from erp-dev/erp
feat: tasks for sync products and customers
This commit is contained in:
330
api_v1/tasks.py
330
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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user