feat: external_order refetch api

api_v1/tasks.py

@@ -16,6 +16,7 @@ import requests
 from celery import shared_task
 from django.conf import settings
 from django.contrib.auth import get_user_model
+from django.db import transaction
 from django.core.files.base import ContentFile
 from django.utils import timezone
 
@@ -39,6 +40,23 @@ logger = logging.getLogger(__name__)
 User = get_user_model()
 
 
+class ExternalPrintingOrderSnapshotSyncError(RuntimeError):
+    def __init__(self, message: str, *, status_code: int = 400, audit_id: int | None = None):
+        super().__init__(message)
+        self.status_code = status_code
+        self.audit_id = audit_id
+
+
+class ExternalPrintingOrderSnapshotNotFoundError(ExternalPrintingOrderSnapshotSyncError):
+    def __init__(self, message: str, *, audit_id: int | None = None):
+        super().__init__(message, status_code=404, audit_id=audit_id)
+
+
+class ExternalPrintingOrderSnapshotConflictError(ExternalPrintingOrderSnapshotSyncError):
+    def __init__(self, message: str, *, audit_id: int | None = None):
+        super().__init__(message, status_code=409, audit_id=audit_id)
+
+
 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')
@@ -308,6 +326,305 @@ def _advance_external_printing_cursor(*, cursor_value: int) -> dict:
     return payload
 
 
+def _fetch_external_printing_order_snapshot(*, external_order_id: str) -> dict:
+    normalized = str(external_order_id or '').strip()
+    if not normalized:
+        raise ExternalPrintingOrderSnapshotSyncError('external_order_id 不能为空', status_code=400)
+
+    url = _build_external_url('/api/v1/records/by-order')
+    try:
+        response = requests.get(
+            url,
+            headers=_get_external_headers(),
+            params={'external_order_id': normalized},
+            timeout=30,
+        )
+    except requests.RequestException as exc:
+        raise ExternalPrintingOrderSnapshotSyncError(
+            f'请求外部订单快照接口失败: {exc}',
+            status_code=502,
+        ) from exc
+
+    if response.status_code == 404:
+        message = f'未找到 external_order_id={normalized} 对应的外部订单'
+        try:
+            payload = response.json()
+        except ValueError:
+            payload = {}
+        if isinstance(payload, dict):
+            message = str(payload.get('message') or message)
+        raise ExternalPrintingOrderSnapshotNotFoundError(message)
+
+    if response.status_code >= 400:
+        message = f'外部订单快照接口返回异常状态码: {response.status_code}'
+        try:
+            payload = response.json()
+        except ValueError:
+            payload = {}
+        if isinstance(payload, dict):
+            message = str(payload.get('message') or payload.get('error') or message)
+        raise ExternalPrintingOrderSnapshotSyncError(message, status_code=502)
+
+    try:
+        payload = response.json()
+    except ValueError as exc:
+        raise ExternalPrintingOrderSnapshotSyncError(
+            '外部订单快照接口返回格式非法：非 JSON object',
+            status_code=502,
+        ) from exc
+
+    if not isinstance(payload, dict):
+        raise ExternalPrintingOrderSnapshotSyncError(
+            '外部订单快照接口返回格式非法：非 JSON object',
+            status_code=502,
+        )
+    return payload
+
+
+def _serialize_datetime_for_snapshot(value):
+    if value is None:
+        return None
+    if hasattr(value, 'isoformat'):
+        return value.isoformat()
+    return str(value)
+
+
+def _get_external_printing_order_for_sync(*, merchant, external_order_id: str):
+    return (
+        printing_models.PrintingOrder.objects.filter(
+            merchant=merchant,
+            external_order_id=external_order_id,
+        )
+        .select_related('customer', 'process', 'created_by')
+        .prefetch_related('printing_jobs__product', 'printing_jobs__business_object')
+        .order_by('id')
+        .first()
+    )
+
+
+def _build_printing_order_before_snapshot(printing_order):
+    if not printing_order:
+        return {'exists': False, 'printing_order': None, 'printing_jobs': []}
+
+    jobs = list(
+        printing_order.printing_jobs.select_related('product', 'business_object').order_by('id')
+    )
+    return {
+        'exists': True,
+        'printing_order': {
+            'id': printing_order.id,
+            'human_id': printing_order.human_id,
+            'customer_id': printing_order.customer_id,
+            'customer_name': getattr(printing_order.customer, 'name', None),
+            'fabric': printing_order.fabric,
+            'width': printing_order.width,
+            'is_urgent': printing_order.is_urgent,
+            'area': printing_order.area,
+            'address': printing_order.address,
+            'fabric_source': printing_order.fabric_source,
+            'is_fabric_received': printing_order.is_fabric_received,
+            'craft': printing_order.craft,
+            'description': printing_order.description,
+            'outgoing_date': _serialize_datetime_for_snapshot(printing_order.outgoing_date),
+            'curve': printing_order.curve,
+            'new_curve': printing_order.new_curve,
+            'position': printing_order.position,
+            'printing_warn': printing_order.printing_warn,
+            'rolling_warn': printing_order.rolling_warn,
+            'production_warn': printing_order.production_warn,
+            'print_count': printing_order.print_count,
+            'is_invalid': printing_order.is_invalid,
+            'process_id': printing_order.process_id,
+            'created_by_id': printing_order.created_by_id,
+            'external_order_id': printing_order.external_order_id,
+            'external_customer_id': printing_order.external_customer_id,
+            'external_customer_name': printing_order.external_customer_name,
+            'external_employee_name': printing_order.external_employee_name,
+            'external_raw': printing_order.external_raw,
+            'created_at': _serialize_datetime_for_snapshot(printing_order.created_at),
+            'updated_at': _serialize_datetime_for_snapshot(printing_order.updated_at),
+        },
+        'printing_jobs': [
+            {
+                'id': job.id,
+                'original_id': job.original_id,
+                'product_id': job.product_id,
+                'product_name': getattr(job.product, 'name', None),
+                'quantity': job.quantity,
+                'unit': job.unit,
+                'size': job.size,
+                'pieces': job.pieces,
+                'description': job.description,
+                'work_state': job.work_state,
+                'is_production_completed': job.is_production_completed,
+                'business_object_id': job.business_object_id,
+                'created_by_id': job.created_by_id,
+                'external_product_name': job.external_product_name,
+                'external_raw': job.external_raw,
+                'created_at': _serialize_datetime_for_snapshot(job.created_at),
+                'updated_at': _serialize_datetime_for_snapshot(job.updated_at),
+            }
+            for job in jobs
+        ],
+    }
+
+
+def _create_printing_order_snapshot_sync_audit(*, printing_order, external_order_id: str, operator_user, allow_reset_stateflow: bool):
+    operator_employee = getattr(operator_user, 'employee', None)
+    return printing_models.PrintingOrderExternalSnapshotSyncAudit.objects.create(
+        printing_order=printing_order,
+        external_order_id=external_order_id,
+        operator_user=operator_user,
+        operator_employee=operator_employee,
+        before_snapshot=_build_printing_order_before_snapshot(printing_order),
+        allow_reset_stateflow=allow_reset_stateflow,
+        is_success=False,
+        failure_reason='',
+    )
+
+
+def _mark_printing_order_snapshot_sync_audit_failure(audit, reason: str, *, printing_order=None):
+    if not audit:
+        return
+    audit.printing_order = printing_order or audit.printing_order
+    audit.is_success = False
+    audit.failure_reason = str(reason or '').strip()
+    audit.save(update_fields=['printing_order', 'is_success', 'failure_reason', 'updated_at'])
+
+
+def _mark_printing_order_snapshot_sync_audit_success(audit, *, printing_order):
+    if not audit:
+        return
+    audit.printing_order = printing_order
+    audit.is_success = True
+    audit.failure_reason = ''
+    audit.save(update_fields=['printing_order', 'is_success', 'failure_reason', 'updated_at'])
+
+
+def _get_printing_order_active_sales_item_info(printing_order):
+    from shipment.services import get_active_sales_items_queryset
+
+    job_ids = list(printing_order.printing_jobs.values_list('id', flat=True))
+    if not job_ids:
+        return {'job_ids': [], 'sales_item_ids': []}
+
+    sales_items = list(
+        get_active_sales_items_queryset()
+        .filter(printing_job_id__in=job_ids)
+        .values('id', 'printing_job_id')
+        .order_by('id')
+    )
+    return {
+        'job_ids': sorted({item['printing_job_id'] for item in sales_items if item['printing_job_id']}),
+        'sales_item_ids': [item['id'] for item in sales_items],
+    }
+
+
+def _get_printing_order_started_job_ids(printing_order):
+    started_job_ids = []
+    for job in printing_order.printing_jobs.select_related('business_object').all():
+        business_object = getattr(job, 'business_object', None)
+        if business_object and business_object.state_logs.filter(is_cancelled=False).exists():
+            started_job_ids.append(job.id)
+    return started_job_ids
+
+
+def _validate_external_printing_order_snapshot_payload(*, payload: dict, external_order_id: str) -> list[dict]:
+    if str(payload.get('mode') or '').strip() != 'snapshot':
+        raise ExternalPrintingOrderSnapshotSyncError('外部订单快照接口返回的 mode 非 snapshot', status_code=502)
+
+    payload_external_order_id = str(payload.get('external_order_id') or '').strip()
+    if payload_external_order_id != external_order_id:
+        raise ExternalPrintingOrderSnapshotSyncError(
+            f'外部订单快照接口返回的 external_order_id 不匹配: {payload_external_order_id or "<empty>"}',
+            status_code=502,
+        )
+
+    status_value = str(payload.get('status') or '').strip()
+    if status_value and status_value != 'active':
+        raise ExternalPrintingOrderSnapshotSyncError(
+            f'外部订单快照状态不支持同步: {status_value}',
+            status_code=409,
+        )
+
+    records = payload.get('records', [])
+    if not isinstance(records, list):
+        raise ExternalPrintingOrderSnapshotSyncError('外部订单快照接口返回格式非法：records 不是数组', status_code=502)
+    if not records:
+        raise ExternalPrintingOrderSnapshotSyncError('外部订单快照接口返回空 records，无法执行覆盖同步', status_code=409)
+
+    for record in records:
+        if not isinstance(record, dict):
+            raise ExternalPrintingOrderSnapshotSyncError('外部订单快照接口返回格式非法：record 不是对象', status_code=502)
+        record_external_order_id = str(record.get('BianHaoID') or '').strip()
+        if record_external_order_id != external_order_id:
+            raise ExternalPrintingOrderSnapshotSyncError(
+                f'外部订单快照中存在不属于目标订单的记录: {record_external_order_id or "<empty>"}',
+                status_code=502,
+            )
+    return records
+
+
+def _reset_printing_order_stateflow_progress(printing_order) -> int:
+    from stateflow.services import reset_business_object_progress
+
+    reset_count = 0
+    for job in printing_order.printing_jobs.select_related('business_object').all():
+        business_object = getattr(job, 'business_object', None)
+        if not business_object:
+            continue
+        if business_object.state_logs.filter(is_cancelled=False).exists():
+            reset_business_object_progress(business_object)
+            reset_count += 1
+    return reset_count
+
+
+def _delete_stale_printing_jobs_for_external_snapshot(*, printing_order, keep_job_ids: set[int]) -> int:
+    from business.models import SalesOrderItem
+
+    stale_jobs = list(
+        printing_order.printing_jobs.select_related('business_object').exclude(id__in=keep_job_ids).order_by('id')
+    )
+    if not stale_jobs:
+        return 0
+
+    stale_job_ids = [job.id for job in stale_jobs]
+    active_sales_items = _get_printing_order_active_sales_item_info(printing_order)
+    blocked_sales_item_job_ids = [job_id for job_id in active_sales_items['job_ids'] if job_id in stale_job_ids]
+    if blocked_sales_item_job_ids:
+        raise ExternalPrintingOrderSnapshotConflictError(
+            '以下待删除 printing_job 已关联销售品，禁止覆盖同步: '
+            f'{blocked_sales_item_job_ids}'
+        )
+
+    blocked_sales_order_job_ids = list(
+        SalesOrderItem.objects.filter(printing_job_id__in=stale_job_ids)
+        .order_by('printing_job_id')
+        .values_list('printing_job_id', flat=True)
+        .distinct()
+    )
+    if blocked_sales_order_job_ids:
+        raise ExternalPrintingOrderSnapshotConflictError(
+            '以下待删除 printing_job 已关联销售单明细，禁止覆盖同步: '
+            f'{blocked_sales_order_job_ids}'
+        )
+
+    deleted_count = 0
+    for job in stale_jobs:
+        business_object = getattr(job, 'business_object', None)
+        has_logs = bool(
+            business_object and business_object.state_logs.exists()
+        )
+        if business_object:
+            job.business_object = None
+            job.save(update_fields=['business_object'])
+        job.delete()
+        if business_object and not has_logs:
+            business_object.delete()
+        deleted_count += 1
+    return deleted_count
+
+
 def _group_external_records(records: list[dict]) -> dict[str, list[dict]]:
     grouped: dict[str, list[dict]] = defaultdict(list)
     for record in records:
@@ -767,6 +1084,134 @@ def _sync_external_printing_records_batch(
     }
 
 
+def sync_external_printing_order_snapshot_impl(
+    *,
+    external_order_id: str,
+    operator_user,
+    allow_reset_stateflow: bool = False,
+) -> dict:
+    normalized_external_order_id = str(external_order_id or '').strip()
+    if not normalized_external_order_id:
+        raise ExternalPrintingOrderSnapshotSyncError('external_order_id 不能为空', status_code=400)
+
+    operator_employee = getattr(operator_user, 'employee', None)
+    merchant = getattr(operator_employee, 'merchant', None)
+    if operator_employee is None or merchant is None:
+        raise ExternalPrintingOrderSnapshotSyncError('当前用户未绑定 employee.merchant，无法执行同步', status_code=403)
+
+    existing_order = _get_external_printing_order_for_sync(
+        merchant=merchant,
+        external_order_id=normalized_external_order_id,
+    )
+    audit = _create_printing_order_snapshot_sync_audit(
+        printing_order=existing_order,
+        external_order_id=normalized_external_order_id,
+        operator_user=operator_user,
+        allow_reset_stateflow=allow_reset_stateflow,
+    )
+
+    try:
+        if existing_order:
+            active_sales_item_info = _get_printing_order_active_sales_item_info(existing_order)
+            if active_sales_item_info['sales_item_ids']:
+                raise ExternalPrintingOrderSnapshotConflictError(
+                    '目标订单存在已关联销售品的 printing_job，禁止覆盖同步。'
+                    f" printing_job_ids={active_sales_item_info['job_ids']},"
+                    f" sales_item_ids={active_sales_item_info['sales_item_ids']}",
+                    audit_id=audit.id,
+                )
+
+            started_job_ids = _get_printing_order_started_job_ids(existing_order)
+            if started_job_ids and not allow_reset_stateflow:
+                raise ExternalPrintingOrderSnapshotConflictError(
+                    '目标订单存在已执行工序的 printing_job，默认不允许覆盖同步。'
+                    ' 如确认要覆盖，请传 allow_reset_stateflow=true。'
+                    f' printing_job_ids={started_job_ids}',
+                    audit_id=audit.id,
+                )
+
+        snapshot_payload = _fetch_external_printing_order_snapshot(
+            external_order_id=normalized_external_order_id,
+        )
+        records = _validate_external_printing_order_snapshot_payload(
+            payload=snapshot_payload,
+            external_order_id=normalized_external_order_id,
+        )
+        sync_user = _get_printing_sync_user()
+        category = _get_printing_sync_product_category(merchant)
+
+        with transaction.atomic():
+            reset_stateflow_job_count = 0
+            if existing_order and allow_reset_stateflow:
+                reset_stateflow_job_count = _reset_printing_order_stateflow_progress(existing_order)
+
+            printing_order, order_created = _upsert_external_printing_order(
+                merchant=merchant,
+                external_order_id=normalized_external_order_id,
+                group_records=records,
+                sync_user=sync_user,
+            )
+
+            jobs_created = 0
+            jobs_updated = 0
+            kept_job_ids: set[int] = set()
+            for record in records:
+                job, job_created = _upsert_external_printing_job(
+                    merchant=merchant,
+                    printing_order=printing_order,
+                    record=record,
+                    sync_user=sync_user,
+                    category=category,
+                )
+                kept_job_ids.add(job.id)
+                if job_created:
+                    jobs_created += 1
+                else:
+                    jobs_updated += 1
+
+            jobs_deleted = _delete_stale_printing_jobs_for_external_snapshot(
+                printing_order=printing_order,
+                keep_job_ids=kept_job_ids,
+            )
+
+        _mark_printing_order_snapshot_sync_audit_success(
+            audit,
+            printing_order=printing_order,
+        )
+        return {
+            'audit_id': audit.id,
+            'external_order_id': normalized_external_order_id,
+            'printing_order_id': printing_order.id,
+            'orders_created': 1 if order_created else 0,
+            'orders_updated': 0 if order_created else 1,
+            'jobs_created': jobs_created,
+            'jobs_updated': jobs_updated,
+            'jobs_deleted': jobs_deleted,
+            'reset_stateflow_job_count': reset_stateflow_job_count,
+        }
+    except ExternalPrintingOrderSnapshotSyncError as exc:
+        if exc.audit_id is None:
+            exc.audit_id = audit.id
+        _mark_printing_order_snapshot_sync_audit_failure(
+            audit,
+            str(exc),
+            printing_order=existing_order,
+        )
+        raise
+    except Exception as exc:
+        reason = f'外部订单快照同步失败: {exc}'
+        _mark_printing_order_snapshot_sync_audit_failure(
+            audit,
+            reason,
+            printing_order=existing_order,
+        )
+        raise ExternalPrintingOrderSnapshotSyncError(
+            reason,
+            status_code=500,
+            audit_id=audit.id,
+        ) from exc
+
+
 def retry_external_printing_sync_failures_impl(
     *,
     limit: int | None = None,