import asyncio
import json
import logging
import os
import shutil
import subprocess
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path

from celery import shared_task
from django.conf import settings
from django.utils import timezone

from basic_info import models as basic_models
from api_v1 import models as api_models
from flower.utils import (
    fetch_products_from_mingdaoyun,
    fetch_customers_from_mingdaoyun,
)
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,
)


logger = logging.getLogger(__name__)


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 _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


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)