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 api_v1.external_product_image_backfill import run_external_product_image_backfill 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']) @shared_task(bind=True) def backfill_external_product_images( self, merchant_id: int | None = None, product_id: int | None = None, limit: int = 500, dry_run: bool = False, ): """ 为来自外部印染订单的产品补图。 仅处理： - printing_order.external_order_id 非空 - product.image 为空 """ payload = run_external_product_image_backfill( fetch_image=_fetch_external_product_image, upload_image=_upload_product_image, merchant_id=merchant_id, product_id=product_id, limit=max(1, int(limit or 500)), dry_run=bool(dry_run), on_success=lambda message: logger.info(message), on_error=lambda message: logger.error(message), on_info=lambda message: logger.info(message), ) payload["task_id"] = self.request.id logger.info("外部印染订单产品补图完成: %s", payload) return payload 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(), 'area': str(first.get('area') 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)