forked from erp-dev/erp
1519 lines
50 KiB
Python
1519 lines
50 KiB
Python
import asyncio
|
||
import base64
|
||
import json
|
||
import logging
|
||
import mimetypes
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
from collections import defaultdict
|
||
from datetime import datetime
|
||
from decimal import Decimal, InvalidOperation
|
||
from pathlib import Path
|
||
|
||
import requests
|
||
from celery import shared_task
|
||
from django.conf import settings
|
||
from django.contrib.auth import get_user_model
|
||
from django.core.files.base import ContentFile
|
||
from django.utils import timezone
|
||
|
||
from basic_info import models as basic_models
|
||
from api_v1 import models as api_models
|
||
from api_v1.views.printing.services import PrintingJobService, PrintingOrderService
|
||
from flower.utils import (
|
||
fetch_products_from_mingdaoyun,
|
||
fetch_customers_from_mingdaoyun,
|
||
)
|
||
from api_v1.mdy_plate_order_sync import sync_mdy_plate_orders_to_staging
|
||
from api_v1.mdy_plate_order_staging_tiia_upload import (
|
||
build_default_tiia_rate_limiter,
|
||
upload_mdy_plate_order_staging_plate_images_to_tencent_tiia,
|
||
)
|
||
from printing import models as printing_models
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
User = get_user_model()
|
||
|
||
|
||
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')
|
||
backup_dir.mkdir(parents=True, exist_ok=True)
|
||
return backup_dir
|
||
|
||
|
||
def _build_backup_path(backup_dir: Path, filename_prefix: str) -> Path:
|
||
timestamp = timezone.now().strftime('%Y%m%d-%H%M%S')
|
||
return backup_dir / f'{filename_prefix}-{timestamp}.sql'
|
||
|
||
|
||
def _run_pg_dump(backup_path: Path):
|
||
db_settings = settings.DATABASES['default']
|
||
pg_dump = shutil.which('pg_dump')
|
||
if not pg_dump:
|
||
raise RuntimeError('pg_dump 不存在，请确认 PostgreSQL 客户端工具已安装')
|
||
|
||
host = db_settings.get('HOST') or 'localhost'
|
||
port = db_settings.get('PORT') or '5432'
|
||
user = db_settings.get('USER') or ''
|
||
name = db_settings['NAME']
|
||
password = db_settings.get('PASSWORD') or ''
|
||
|
||
cmd = [
|
||
pg_dump,
|
||
'-h',
|
||
host,
|
||
'-p',
|
||
str(port),
|
||
'-U',
|
||
user,
|
||
'-F',
|
||
'p',
|
||
'-d',
|
||
name,
|
||
]
|
||
|
||
env = os.environ.copy()
|
||
if password:
|
||
env['PGPASSWORD'] = password
|
||
|
||
with backup_path.open('wb') as stream:
|
||
subprocess.run(cmd, check=True, stdout=stream, env=env)
|
||
|
||
|
||
def _dump_database_to_sql(backup_path: Path):
|
||
engine = settings.DATABASES['default']['ENGINE']
|
||
if 'postgresql' not in engine:
|
||
raise NotImplementedError('当前项目只支持 PostgreSQL 数据库备份，请检查 DATABASES 配置')
|
||
_run_pg_dump(backup_path)
|
||
|
||
|
||
@shared_task(bind=True)
|
||
def backup_database(self, output_dir: str | None = None, filename_prefix: str = 'db-backup'):
|
||
"""
|
||
备份当前数据库为 .sql 文件（仅数据，不包含表结构 DDL），存放在项目根目录 data-bak 下。
|
||
|
||
参数:
|
||
output_dir: 可选，指定备份目录（默认 BASE_DIR/data-bak）
|
||
filename_prefix: 备份文件名前缀
|
||
"""
|
||
backup_dir = _ensure_backup_dir(output_dir)
|
||
backup_path = _build_backup_path(backup_dir, filename_prefix)
|
||
_dump_database_to_sql(backup_path)
|
||
|
||
payload = {
|
||
'task_id': self.request.id,
|
||
'backup_path': str(backup_path),
|
||
'created_at': timezone.now().isoformat(),
|
||
}
|
||
logger.info('数据库备份完成: %s', payload)
|
||
return payload
|
||
|
||
|
||
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)
|
||
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,
|
||
'段': basic_models.ProductUnitEnum.SEGMENT,
|
||
}
|
||
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 _ensure_int(value):
|
||
if value is None:
|
||
return None
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError, InvalidOperation):
|
||
# 有些值可能是 Decimal 或字符串数字
|
||
try:
|
||
return int(Decimal(str(value)))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _parse_external_datetime(value: str | None):
|
||
if not value:
|
||
return None
|
||
text = str(value).strip()
|
||
if not text:
|
||
return None
|
||
try:
|
||
dt = datetime.fromisoformat(text.replace('Z', '+00:00'))
|
||
except ValueError:
|
||
return None
|
||
if timezone.is_naive(dt):
|
||
dt = timezone.make_aware(dt, timezone.get_current_timezone())
|
||
return dt
|
||
|
||
|
||
def _extract_positive_int(value, *, field_name: str, allow_blank: bool = True):
|
||
if value is None or value == '':
|
||
return None if allow_blank else 0
|
||
try:
|
||
number = Decimal(str(value))
|
||
except (InvalidOperation, TypeError, ValueError) as exc:
|
||
raise RuntimeError(f'外部字段 {field_name} 无法解析为整数: {value!r}') from exc
|
||
if number != number.to_integral_value():
|
||
raise RuntimeError(f'外部字段 {field_name} 不是整数: {value!r}')
|
||
parsed = int(number)
|
||
if parsed < 0:
|
||
raise RuntimeError(f'外部字段 {field_name} 不能为负数: {value!r}')
|
||
return parsed
|
||
|
||
|
||
def _extract_pieces(value):
|
||
if value is None or value == '':
|
||
return None
|
||
match = re.search(r'(\d+)', str(value))
|
||
if not match:
|
||
return None
|
||
return int(match.group(1))
|
||
|
||
|
||
def _normalize_unit(value):
|
||
text = str(value or '').strip()
|
||
return text.lstrip('/').strip() or text
|
||
|
||
|
||
def _split_fabric_source_and_craft(value: str | None):
|
||
text = str(value or '').strip()
|
||
if not text:
|
||
return '', ''
|
||
parts = re.split(r'\s+', text)
|
||
fabric_source = parts[0] if parts else ''
|
||
craft = ' '.join(parts[1:]).strip() if len(parts) > 1 else ''
|
||
return fabric_source, craft
|
||
|
||
|
||
def _build_external_url(path: str) -> str:
|
||
base_url = str(getattr(settings, 'PRINTING_EXTERNAL_RECORDS_BASE_URL', '') or '').rstrip('/')
|
||
if not base_url:
|
||
raise RuntimeError('未配置 PRINTING_EXTERNAL_RECORDS_BASE_URL')
|
||
return f'{base_url}{path}'
|
||
|
||
|
||
def _get_external_headers() -> dict[str, str]:
|
||
secret = str(getattr(settings, 'PRINTING_EXTERNAL_RECORDS_AUTHORIZATION', '') or '').strip()
|
||
if not secret:
|
||
raise RuntimeError('未配置 PRINTING_EXTERNAL_RECORDS_AUTHORIZATION')
|
||
return {'Authorization': secret}
|
||
|
||
|
||
def _get_printing_sync_user():
|
||
user_id = int(getattr(settings, 'PRINTING_EXTERNAL_SYNC_USER_ID', 1) or 1)
|
||
user = User.objects.filter(id=user_id).select_related('employee__merchant').first()
|
||
if not user:
|
||
raise RuntimeError(f'未找到固定同步用户: {user_id}')
|
||
employee = getattr(user, 'employee', None)
|
||
merchant = getattr(employee, 'merchant', None)
|
||
if not merchant:
|
||
raise RuntimeError(f'固定同步用户 {user_id} 未绑定 employee.merchant，无法确定同步商户')
|
||
return user
|
||
|
||
|
||
def _get_printing_sync_product_category(merchant):
|
||
category_id = getattr(settings, 'PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID', None)
|
||
qs = basic_models.ProductCategory.objects.filter(merchant=merchant).order_by('id')
|
||
if category_id:
|
||
qs = qs.filter(id=category_id)
|
||
category = qs.first()
|
||
if not category:
|
||
raise RuntimeError(
|
||
f'未找到用于外部印染同步的产品类别，请先创建类别或配置 '
|
||
f'PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID（merchant={merchant.id}）'
|
||
)
|
||
return category
|
||
|
||
|
||
def _fetch_external_printing_records(*, limit: int = 100) -> dict:
|
||
url = _build_external_url('/api/v1/records')
|
||
response = requests.get(
|
||
url,
|
||
headers=_get_external_headers(),
|
||
params={'limit': max(1, int(limit)), 'update_cursor': 'false'},
|
||
timeout=30,
|
||
)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
if not isinstance(payload, dict):
|
||
raise RuntimeError('外部 records 接口返回格式非法：非 JSON object')
|
||
if not isinstance(payload.get('records', []), list):
|
||
raise RuntimeError('外部 records 接口返回格式非法：records 不是数组')
|
||
return payload
|
||
|
||
|
||
def _advance_external_printing_cursor(*, cursor_value: int) -> dict:
|
||
url = _build_external_url('/api/v1/cursor/set')
|
||
response = requests.post(
|
||
url,
|
||
headers=_get_external_headers(),
|
||
json={'value': int(cursor_value)},
|
||
timeout=30,
|
||
)
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
if not isinstance(payload, dict):
|
||
raise RuntimeError('外部 cursor/set 接口返回格式非法：非 JSON object')
|
||
return payload
|
||
|
||
|
||
def _group_external_records(records: list[dict]) -> dict[str, list[dict]]:
|
||
grouped: dict[str, list[dict]] = defaultdict(list)
|
||
for record in records:
|
||
external_order_id = str(record.get('BianHaoID') or '').strip()
|
||
if not external_order_id:
|
||
raise RuntimeError(f'外部 record 缺少 BianHaoID: {record!r}')
|
||
grouped[external_order_id].append(record)
|
||
return grouped
|
||
|
||
|
||
def _resolve_external_customer(*, merchant, record: dict):
|
||
customer_payload = record.get('customer') or {}
|
||
customer_name = str(customer_payload.get('KhName') or '').strip()
|
||
external_customer_id = str(record.get('KhID') or customer_payload.get('KhID') or '').strip()
|
||
if not customer_name:
|
||
raise RuntimeError(f'外部 record 缺少 customer.KhName: {record!r}')
|
||
|
||
customer = (
|
||
basic_models.Customer.objects.filter(merchant=merchant, name=customer_name)
|
||
.order_by('id')
|
||
.first()
|
||
)
|
||
if not customer:
|
||
defaults = {
|
||
'merchant': merchant,
|
||
'name': customer_name,
|
||
}
|
||
sync_user = record.get('_sync_user')
|
||
employee = getattr(sync_user, 'employee', None)
|
||
if employee and employee.merchant_id == merchant.id:
|
||
defaults['created_by'] = employee
|
||
customer = basic_models.Customer.objects.create(**defaults)
|
||
logger.info(
|
||
'外部印染同步自动创建客户: merchant_id=%s customer_name=%s external_customer_id=%s customer_id=%s',
|
||
merchant.id,
|
||
customer_name,
|
||
external_customer_id,
|
||
customer.id,
|
||
)
|
||
return customer, customer_name, external_customer_id
|
||
|
||
|
||
def _resolve_external_employee_user(*, merchant, external_employee_name: str, sync_user):
|
||
normalized = str(external_employee_name or '').strip()
|
||
if not normalized:
|
||
return sync_user, ''
|
||
|
||
employee = (
|
||
basic_models.Employee.objects.filter(merchant=merchant, name=normalized)
|
||
.select_related('sys_user')
|
||
.order_by('id')
|
||
.first()
|
||
)
|
||
if employee and employee.sys_user_id:
|
||
return employee.sys_user, ''
|
||
return sync_user, normalized
|
||
|
||
|
||
def _encode_external_image_name_b64(name: str) -> str:
|
||
raw = str(name or '').encode('utf-8')
|
||
return base64.urlsafe_b64encode(raw).decode('ascii').rstrip('=')
|
||
|
||
|
||
def _decode_external_image_data(data: str) -> bytes:
|
||
text = str(data or '').strip()
|
||
if not text:
|
||
raise RuntimeError('外部图片接口返回空 data')
|
||
padding = '=' * (-len(text) % 4)
|
||
try:
|
||
return base64.b64decode(text + padding)
|
||
except Exception as exc:
|
||
raise RuntimeError('外部图片接口返回的 data 不是合法 base64') from exc
|
||
|
||
|
||
def _fetch_external_product_image(external_product_name: str) -> dict:
|
||
url = _build_external_url('/api/v1/image')
|
||
response = requests.get(
|
||
url,
|
||
headers=_get_external_headers(),
|
||
params={'name_b64': _encode_external_image_name_b64(external_product_name)},
|
||
timeout=30,
|
||
)
|
||
if response.status_code == 404:
|
||
raise RuntimeError(f'外部图片不存在: {external_product_name}')
|
||
response.raise_for_status()
|
||
payload = response.json()
|
||
if not isinstance(payload, dict):
|
||
raise RuntimeError('外部图片接口返回格式非法：非 JSON object')
|
||
data = _decode_external_image_data(payload.get('data', ''))
|
||
content_type = str(payload.get('content_type') or '').strip() or 'application/octet-stream'
|
||
image_name = str(payload.get('name') or '').strip()
|
||
return {
|
||
'bytes': data,
|
||
'content_type': content_type,
|
||
'name': image_name or external_product_name,
|
||
}
|
||
|
||
|
||
def _build_product_image_filename(product_name: str, image_name: str, content_type: str) -> str:
|
||
suffix = Path(str(image_name or '').strip()).suffix
|
||
if not suffix:
|
||
suffix = mimetypes.guess_extension(content_type or '') or '.jpg'
|
||
safe_name = re.sub(r'[^0-9A-Za-z._-]+', '_', str(product_name or '').strip()) or 'external_product'
|
||
return f'{safe_name}{suffix}'
|
||
|
||
|
||
def _upload_product_image(product, image_payload: dict):
|
||
filename = _build_product_image_filename(
|
||
product_name=product.name,
|
||
image_name=image_payload.get('name', ''),
|
||
content_type=image_payload.get('content_type', ''),
|
||
)
|
||
content = ContentFile(image_payload['bytes'])
|
||
product.image.save(filename, content, save=False)
|
||
product.save(update_fields=['image', 'updated_at'])
|
||
|
||
|
||
def _record_printing_external_sync_failure(*, run_date, record: dict, error: str):
|
||
external_record_id = _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
|
||
obj, _created = api_models.PrintingExternalSyncFailure.objects.get_or_create(
|
||
run_date=run_date,
|
||
external_record_id=external_record_id,
|
||
defaults={
|
||
'external_order_id': str(record.get('BianHaoID') or '').strip(),
|
||
'product_name': str(record.get('YanSe') or '').strip(),
|
||
'error': error or '',
|
||
'raw': record,
|
||
'attempts': 0,
|
||
'last_attempt_at': timezone.now(),
|
||
},
|
||
)
|
||
obj.external_order_id = str(record.get('BianHaoID') or '').strip()
|
||
obj.product_name = str(record.get('YanSe') or '').strip()
|
||
obj.error = error or ''
|
||
obj.raw = record
|
||
obj.attempts = int(obj.attempts or 0) + 1
|
||
obj.last_attempt_at = timezone.now()
|
||
obj.save(
|
||
update_fields=[
|
||
'external_order_id',
|
||
'product_name',
|
||
'error',
|
||
'raw',
|
||
'attempts',
|
||
'last_attempt_at',
|
||
]
|
||
)
|
||
|
||
|
||
def _ensure_external_product(*, merchant, external_product_name: str, category):
|
||
normalized = str(external_product_name or '').strip()
|
||
if not normalized:
|
||
raise RuntimeError('外部产品名为空，无法创建/绑定产品')
|
||
|
||
existing_product = (
|
||
basic_models.Product.objects.filter(merchant=merchant, name=normalized)
|
||
.order_by('id')
|
||
.first()
|
||
)
|
||
if existing_product:
|
||
return existing_product, False
|
||
|
||
image_payload = _fetch_external_product_image(normalized)
|
||
product = basic_models.Product.objects.create(
|
||
merchant=merchant,
|
||
category=category,
|
||
name=normalized,
|
||
)
|
||
try:
|
||
_upload_product_image(product, image_payload)
|
||
except Exception:
|
||
if product.image:
|
||
try:
|
||
product.image.delete(save=False)
|
||
except Exception:
|
||
logger.warning('回滚外部产品图片失败: product_id=%s', product.id, exc_info=True)
|
||
product.delete()
|
||
raise
|
||
return product, True
|
||
|
||
|
||
def _build_order_external_raw(group_records: list[dict]) -> dict:
|
||
first = group_records[0] if group_records else {}
|
||
return {
|
||
'group_size': len(group_records),
|
||
'record_ids': [record.get('ID') for record in group_records],
|
||
'first_record': first,
|
||
}
|
||
|
||
|
||
def _build_external_order_data(
|
||
*,
|
||
merchant,
|
||
external_order_id: str,
|
||
group_records: list[dict],
|
||
sync_user,
|
||
):
|
||
first = group_records[0]
|
||
first_with_context = dict(first)
|
||
first_with_context['_sync_user'] = sync_user
|
||
customer, external_customer_name, external_customer_id = _resolve_external_customer(
|
||
merchant=merchant,
|
||
record=first_with_context,
|
||
)
|
||
created_by_user, external_employee_name = _resolve_external_employee_user(
|
||
merchant=merchant,
|
||
external_employee_name=str(first.get('CaoZY') or '').strip(),
|
||
sync_user=sync_user,
|
||
)
|
||
fabric_source, craft = _split_fabric_source_and_craft(first.get('SHDZ'))
|
||
return {
|
||
'merchant': merchant,
|
||
'customer': customer,
|
||
'fabric': str(first.get('HpName') or '').strip(),
|
||
'width': str(first.get('SeHao') or '').strip(),
|
||
'fabric_source': fabric_source or None,
|
||
'craft': craft or None,
|
||
'rolling_warn': str(first.get('BeiZhu') or '').strip() or None,
|
||
'curve': str(first.get('MeoA') or '').strip() or None,
|
||
'position': str(first.get('FidJ') or '').strip() or None,
|
||
'outgoing_date': _parse_external_datetime(first.get('KdRiQi')),
|
||
'created_by': created_by_user,
|
||
'external_order_id': external_order_id,
|
||
'external_customer_id': external_customer_id or None,
|
||
'external_customer_name': external_customer_name or None,
|
||
'external_employee_name': external_employee_name or None,
|
||
'external_raw': _build_order_external_raw(group_records),
|
||
}
|
||
|
||
|
||
def _upsert_external_printing_order(*, merchant, external_order_id: str, group_records: list[dict], sync_user):
|
||
order_data = _build_external_order_data(
|
||
merchant=merchant,
|
||
external_order_id=external_order_id,
|
||
group_records=group_records,
|
||
sync_user=sync_user,
|
||
)
|
||
created_by_user = order_data['created_by']
|
||
existing_order = (
|
||
printing_models.PrintingOrder.objects.filter(
|
||
merchant=merchant,
|
||
external_order_id=external_order_id,
|
||
)
|
||
.order_by('id')
|
||
.first()
|
||
)
|
||
if existing_order:
|
||
success, message, order = PrintingOrderService.update_printing_order(
|
||
existing_order,
|
||
order_data,
|
||
created_by_user,
|
||
)
|
||
if not success:
|
||
raise RuntimeError(f'更新印染订单失败: {message}')
|
||
return order, False
|
||
order = PrintingOrderService.create_printing_order(order_data, created_by_user)
|
||
return order, True
|
||
|
||
|
||
def _upsert_external_printing_job(*, merchant, printing_order, record: dict, sync_user, category):
|
||
record_id = _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
|
||
external_product_name = str(record.get('YanSe') or '').strip()
|
||
created_by_user, _external_employee_name = _resolve_external_employee_user(
|
||
merchant=merchant,
|
||
external_employee_name=str(record.get('CaoZY') or '').strip(),
|
||
sync_user=sync_user,
|
||
)
|
||
product, _ = _ensure_external_product(
|
||
merchant=merchant,
|
||
external_product_name=external_product_name,
|
||
category=category,
|
||
)
|
||
job_data = {
|
||
'merchant': merchant,
|
||
'printing_order': printing_order,
|
||
'product': product,
|
||
'quantity': _extract_positive_int(record.get('ShuLiang'), field_name='ShuLiang', allow_blank=False),
|
||
'unit': _normalize_unit(record.get('JiJiaDW')),
|
||
'size': str(record.get('ShuLiangZ') or '').strip() or None,
|
||
'pieces': _extract_pieces(record.get('BeiZhuC')),
|
||
'description': None,
|
||
'original_id': record_id,
|
||
'created_by': created_by_user,
|
||
'external_product_name': None,
|
||
'external_raw': record,
|
||
}
|
||
existing_job = (
|
||
printing_models.PrintingJob.objects.filter(
|
||
printing_order=printing_order,
|
||
product=product,
|
||
)
|
||
.order_by('id')
|
||
.first()
|
||
)
|
||
if existing_job:
|
||
success, message, job = PrintingJobService.update_printing_job(
|
||
existing_job,
|
||
job_data,
|
||
created_by_user,
|
||
)
|
||
if not success:
|
||
raise RuntimeError(f'更新印染任务失败: {message}')
|
||
return job, False
|
||
job = PrintingJobService.create_printing_job(job_data, created_by_user)
|
||
return job, True
|
||
|
||
|
||
def _sync_external_printing_records_batch(
|
||
*,
|
||
records: list[dict],
|
||
merchant,
|
||
sync_user,
|
||
category,
|
||
record_failure,
|
||
) -> dict:
|
||
orders_created = 0
|
||
orders_updated = 0
|
||
jobs_created = 0
|
||
jobs_updated = 0
|
||
failed_record_ids: list[int] = []
|
||
failed_records = 0
|
||
grouped_records: dict[str, list[dict]] = {}
|
||
|
||
if not records:
|
||
return {
|
||
'orders_created': 0,
|
||
'orders_updated': 0,
|
||
'jobs_created': 0,
|
||
'jobs_updated': 0,
|
||
'failed_records': 0,
|
||
'failed_record_ids': [],
|
||
'group_count': 0,
|
||
}
|
||
|
||
grouped_records = _group_external_records(records)
|
||
for external_order_id, group_records in grouped_records.items():
|
||
successful_records: list[dict] = []
|
||
for record in group_records:
|
||
try:
|
||
_ensure_external_product(
|
||
merchant=merchant,
|
||
external_product_name=str(record.get('YanSe') or '').strip(),
|
||
category=category,
|
||
)
|
||
except Exception as exc:
|
||
error_message = str(exc)
|
||
logger.warning(
|
||
'外部印染记录产品处理失败: external_record_id=%s external_order_id=%s error=%s',
|
||
record.get('ID'),
|
||
external_order_id,
|
||
error_message,
|
||
)
|
||
record_failure(record=record, error=error_message)
|
||
failed_records += 1
|
||
failed_record_ids.append(
|
||
_extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
|
||
)
|
||
continue
|
||
successful_records.append(record)
|
||
|
||
if not successful_records:
|
||
continue
|
||
|
||
try:
|
||
order, order_created = _upsert_external_printing_order(
|
||
merchant=merchant,
|
||
external_order_id=external_order_id,
|
||
group_records=successful_records,
|
||
sync_user=sync_user,
|
||
)
|
||
if order_created:
|
||
orders_created += 1
|
||
else:
|
||
orders_updated += 1
|
||
except Exception as exc:
|
||
error_message = str(exc)
|
||
logger.warning(
|
||
'外部印染记录订单处理失败: external_order_id=%s error=%s',
|
||
external_order_id,
|
||
error_message,
|
||
)
|
||
for record in successful_records:
|
||
record_failure(record=record, error=error_message)
|
||
failed_records += 1
|
||
failed_record_ids.append(
|
||
_extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
|
||
)
|
||
continue
|
||
|
||
for record in successful_records:
|
||
try:
|
||
_job, job_created = _upsert_external_printing_job(
|
||
merchant=merchant,
|
||
printing_order=order,
|
||
record=record,
|
||
sync_user=sync_user,
|
||
category=category,
|
||
)
|
||
if job_created:
|
||
jobs_created += 1
|
||
else:
|
||
jobs_updated += 1
|
||
except Exception as exc:
|
||
error_message = str(exc)
|
||
logger.warning(
|
||
'外部印染记录任务处理失败: external_record_id=%s external_order_id=%s error=%s',
|
||
record.get('ID'),
|
||
external_order_id,
|
||
error_message,
|
||
)
|
||
record_failure(record=record, error=error_message)
|
||
failed_records += 1
|
||
failed_record_ids.append(
|
||
_extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
|
||
)
|
||
|
||
return {
|
||
'orders_created': orders_created,
|
||
'orders_updated': orders_updated,
|
||
'jobs_created': jobs_created,
|
||
'jobs_updated': jobs_updated,
|
||
'failed_records': failed_records,
|
||
'failed_record_ids': failed_record_ids,
|
||
'group_count': len(grouped_records),
|
||
}
|
||
|
||
|
||
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
|
||
pieces_int = _ensure_int(product_data.pieces)
|
||
if pieces_int is not None:
|
||
defaults['pieces'] = pieces_int
|
||
segment_decimal = _ensure_decimal(product_data.segment_size)
|
||
if segment_decimal is not None:
|
||
defaults['segment_size'] = segment_decimal
|
||
|
||
# 使用 filter().first() 替代 get_or_create，容忍重复数据
|
||
product_obj = basic_models.Product.objects.filter(
|
||
merchant=merchant,
|
||
human_id=product_data.uid,
|
||
).first()
|
||
|
||
if product_obj is None:
|
||
# 不存在，创建新记录
|
||
product_obj = basic_models.Product.objects.create(
|
||
merchant=merchant,
|
||
human_id=product_data.uid,
|
||
**defaults,
|
||
)
|
||
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 | None = None,
|
||
max_records: int | None = None,
|
||
):
|
||
"""
|
||
从明道云同步产品数据（按 ctime 升序分页扫描）。
|
||
|
||
- 默认从上次同步记录的 page_index 继续翻页
|
||
- last_ctime/last_rowid 用于页内游标（避免重复处理）
|
||
- max_pages 表示“单次任务最多处理多少页”（不是最大页码）
|
||
"""
|
||
merchant = _get_mdy_merchant()
|
||
category = _get_mdy_category(merchant)
|
||
max_records = max_records or 0 # 0 表示不限制
|
||
|
||
last_sync = api_models.DataSync.objects.filter(
|
||
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 ''
|
||
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:
|
||
# max_pages: 单次任务最多处理多少页（不是“最大页码”）
|
||
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:
|
||
# 同一秒内可能有多条记录，尽量把 rowid 推进到最后处理的那条
|
||
latest_rowid = item.rowid
|
||
|
||
pages_processed += 1
|
||
if hit_max_records:
|
||
# 达到单次任务的记录上限：下次从同一页继续（依赖 last_ctime/last_rowid 跳过已处理部分）
|
||
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=None,
|
||
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',
|
||
)
|
||
|
||
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 | None = None,
|
||
max_records: int | None = None,
|
||
):
|
||
"""
|
||
从明道云同步客户数据（按 ctime 升序分页扫描）。
|
||
|
||
- 默认从上次同步记录的 page_index 继续翻页
|
||
- last_ctime/last_rowid 用于页内游标（避免重复处理）
|
||
- max_pages 表示“单次任务最多处理多少页”（不是最大页码）
|
||
"""
|
||
merchant = _get_mdy_merchant()
|
||
max_records = max_records or 0 # 0 表示不限制
|
||
|
||
last_sync = api_models.DataSync.objects.filter(
|
||
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 ''
|
||
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:
|
||
# max_pages: 单次任务最多处理多少页（不是“最大页码”）
|
||
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=None,
|
||
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',
|
||
)
|
||
|
||
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
|
||
|
||
|
||
@shared_task(bind=True)
|
||
def sync_mdy_plate_orders(
|
||
self,
|
||
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=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
|
||
|
||
|
||
@shared_task(bind=True)
|
||
def sync_external_printing_records(self, limit: int = 100):
|
||
"""
|
||
从外部 records 接口增量同步印染订单/任务。
|
||
|
||
策略：
|
||
- 拉取时强制 update_cursor=false
|
||
- 按外部 record.ID 处理失败并记录
|
||
- 本批处理结束后无论是否有部分失败，都推进远端 cursor
|
||
"""
|
||
limit = max(1, int(limit or 100))
|
||
sync_user = _get_printing_sync_user()
|
||
merchant = sync_user.employee.merchant
|
||
category = _get_printing_sync_product_category(merchant)
|
||
payload = _fetch_external_printing_records(limit=limit)
|
||
records = payload.get('records', [])
|
||
last_record_id = payload.get('last_record_id')
|
||
run_date = timezone.localdate()
|
||
|
||
if records:
|
||
batch_result = _sync_external_printing_records_batch(
|
||
records=records,
|
||
merchant=merchant,
|
||
sync_user=sync_user,
|
||
category=category,
|
||
record_failure=lambda *, record, error: _record_printing_external_sync_failure(
|
||
run_date=run_date,
|
||
record=record,
|
||
error=error,
|
||
),
|
||
)
|
||
|
||
cursor_payload = None
|
||
if last_record_id is not None:
|
||
cursor_payload = _advance_external_printing_cursor(cursor_value=int(last_record_id))
|
||
else:
|
||
batch_result = {
|
||
'orders_created': 0,
|
||
'orders_updated': 0,
|
||
'jobs_created': 0,
|
||
'jobs_updated': 0,
|
||
'failed_records': 0,
|
||
'failed_record_ids': [],
|
||
'group_count': 0,
|
||
}
|
||
cursor_payload = None
|
||
|
||
api_models.DataSync.objects.create(
|
||
table_name=api_models.DataSync.TableName.PRINTING_EXTERNAL_RECORD,
|
||
merchant=merchant,
|
||
page_index=1,
|
||
page_size=limit,
|
||
synced_rows=len(records),
|
||
total_count=int(payload.get('count') or 0),
|
||
note=f'cursor_after={payload.get("cursor_after")} last_record_id={last_record_id}',
|
||
)
|
||
|
||
result = {
|
||
'task_id': self.request.id,
|
||
'limit': limit,
|
||
'mode': payload.get('mode'),
|
||
'count': len(records),
|
||
'orders_created': batch_result['orders_created'],
|
||
'orders_updated': batch_result['orders_updated'],
|
||
'jobs_created': batch_result['jobs_created'],
|
||
'jobs_updated': batch_result['jobs_updated'],
|
||
'failed_records': batch_result['failed_records'],
|
||
'failed_record_ids': batch_result['failed_record_ids'],
|
||
'last_record_id': last_record_id,
|
||
'cursor_before': payload.get('cursor_before'),
|
||
'cursor_after': payload.get('cursor_after'),
|
||
'cursor_updated': bool(cursor_payload),
|
||
'group_count': batch_result['group_count'],
|
||
}
|
||
logger.info('外部印染 records 同步完成: %s', result)
|
||
return result
|
||
|
||
|
||
def _record_mdy_plate_order_staging_tiia_failure(
|
||
*,
|
||
run_date,
|
||
staging: api_models.MDYPlateOrderStaging,
|
||
error: str,
|
||
details: list[dict] | None = None,
|
||
) -> None:
|
||
"""
|
||
记录“开版暂存 -> TIIA 上传”失败（同一天同一 mdy_rowid 去重，attempts 累加）。
|
||
"""
|
||
obj, _created = api_models.MDYPlateOrderStagingTiiaUploadFailure.objects.get_or_create(
|
||
run_date=run_date,
|
||
mdy_rowid=staging.mdy_rowid,
|
||
defaults={
|
||
"staging_id": staging.id,
|
||
"error": error or "",
|
||
"details": details or [],
|
||
"attempts": 0,
|
||
"last_attempt_at": timezone.now(),
|
||
},
|
||
)
|
||
obj.staging_id = staging.id
|
||
obj.error = error or ""
|
||
obj.details = details or []
|
||
obj.attempts = int(obj.attempts or 0) + 1
|
||
obj.last_attempt_at = timezone.now()
|
||
obj.save(update_fields=["staging_id", "error", "details", "attempts", "last_attempt_at", "updated_at"])
|
||
|
||
|
||
@shared_task(bind=True)
|
||
def upload_mdy_plate_order_staging_images_to_tencent_tiia(
|
||
self,
|
||
batch_size: int = 200,
|
||
*,
|
||
dry_run: bool = False,
|
||
):
|
||
"""
|
||
将 MDYPlateOrderStaging 的“开版图（Attachment）”上传到腾讯云 TIIA 图库（增量）。
|
||
|
||
- 游标：使用 api_data_sync.last_rowid 存储 staging 表的 last_id（自增主键）
|
||
- 失败：写入 api_mdy_plate_order_staging_tiia_upload_failure（不影响后续继续跑）
|
||
- 限流：按 settings.TENCENTCLOUD_TIIA_QPS（默认 10 qps）
|
||
"""
|
||
run_date = timezone.localdate()
|
||
limiter = build_default_tiia_rate_limiter()
|
||
|
||
last_sync = (
|
||
api_models.DataSync.objects.filter(
|
||
table_name=api_models.DataSync.TableName.MDY_PLATE_ORDER_STAGING_TIIA_UPLOAD
|
||
)
|
||
.order_by("-created_at")
|
||
.first()
|
||
)
|
||
last_id = 0
|
||
if last_sync and (last_sync.last_rowid or "").strip():
|
||
try:
|
||
last_id = int(last_sync.last_rowid)
|
||
except ValueError:
|
||
last_id = 0
|
||
|
||
qs = api_models.MDYPlateOrderStaging.objects.filter(id__gt=last_id).order_by("id")
|
||
if batch_size and batch_size > 0:
|
||
qs = qs[:batch_size]
|
||
|
||
processed = 0
|
||
failed_records = 0
|
||
latest_id = last_id
|
||
latest_ctime = last_sync.last_ctime if last_sync else None
|
||
|
||
for staging in qs:
|
||
processed += 1
|
||
latest_id = staging.id
|
||
if staging.ctime:
|
||
latest_ctime = staging.ctime
|
||
|
||
try:
|
||
result = upload_mdy_plate_order_staging_plate_images_to_tencent_tiia(
|
||
staging=staging,
|
||
rate_limiter=limiter,
|
||
dry_run=dry_run,
|
||
)
|
||
if int(result.get("failure_count") or 0) > 0:
|
||
failed_records += 1
|
||
_record_mdy_plate_order_staging_tiia_failure(
|
||
run_date=run_date,
|
||
staging=staging,
|
||
error="部分图片上传失败" if int(result.get("success_count") or 0) > 0 else "图片上传失败",
|
||
details=[{"result": result}],
|
||
)
|
||
except Exception as exc:
|
||
failed_records += 1
|
||
_record_mdy_plate_order_staging_tiia_failure(
|
||
run_date=run_date,
|
||
staging=staging,
|
||
error=str(exc),
|
||
details=[{"error": str(exc), "mdy_rowid": staging.mdy_rowid, "staging_id": staging.id}],
|
||
)
|
||
|
||
api_models.DataSync.objects.create(
|
||
table_name=api_models.DataSync.TableName.MDY_PLATE_ORDER_STAGING_TIIA_UPLOAD,
|
||
page_index=1,
|
||
page_size=batch_size,
|
||
synced_rows=processed,
|
||
total_count=api_models.MDYPlateOrderStaging.objects.count(),
|
||
last_ctime=latest_ctime,
|
||
last_rowid=str(latest_id) if latest_id else str(last_id),
|
||
note=f"id scan; dry_run={dry_run}",
|
||
)
|
||
|
||
payload = {
|
||
"task_id": self.request.id,
|
||
"processed": processed,
|
||
"failed_records": failed_records,
|
||
"last_id": latest_id,
|
||
"last_ctime": latest_ctime.isoformat() if latest_ctime else None,
|
||
"dry_run": dry_run,
|
||
"batch_size": batch_size,
|
||
}
|
||
logger.info("MDY 开版暂存图片上传到 TIIA 完成: %s", payload)
|
||
return payload
|
||
|
||
|
||
@shared_task
|
||
def save_api_audit_log(
|
||
url: str,
|
||
method: str,
|
||
request_data: dict,
|
||
query_params: dict,
|
||
user_id: int | None,
|
||
username: str,
|
||
response_status: int | None,
|
||
):
|
||
"""
|
||
异步保存API审计日志。
|
||
|
||
通过Celery队列异步执行，避免阻塞API响应。
|
||
|
||
参数:
|
||
url: 请求URL路径
|
||
method: HTTP方法（POST等）
|
||
request_data: 请求体数据（JSON格式）
|
||
query_params: URL查询参数
|
||
user_id: 操作用户ID
|
||
username: 操作用户的用户名
|
||
response_status: HTTP响应状态码
|
||
"""
|
||
try:
|
||
api_models.ApiAuditLog.objects.create(
|
||
url=url,
|
||
method=method,
|
||
request_data=request_data,
|
||
query_params=query_params,
|
||
user_id=user_id,
|
||
username=username,
|
||
response_status=response_status,
|
||
)
|
||
logger.debug('API审计日志已保存: %s %s (user=%s, status=%s)', method, url, username, response_status)
|
||
except Exception as e:
|
||
logger.error('保存API审计日志失败: %s', str(e), exc_info=True)
|