from __future__ import annotations

import hashlib
import json
import logging
from typing import Any, Callable, Dict, List, Optional, Sequence, Set

logger = logging.getLogger(__name__)

DEFAULT_ORPHAN_CLEANUP_BATCH_SIZE = 250
ProgressCallback = Callable[[Dict[str, Any]], None]


def _canonical_json(obj: Any) -> str:
    if obj is None:
        return "null"
    return json.dumps(obj, sort_keys=True, default=str)


def _media_hash(media_files: Sequence[Dict[str, Any]]) -> str:
    return hashlib.sha256(_canonical_json(list(media_files)).encode("utf-8")).hexdigest()


def _build_delete_targets(
    stale_rows: Sequence[Dict[str, Any]],
    *,
    current_by_id: Dict[int, Dict[str, Any]] | None = None,
    current_files: Set[str] | None = None,
) -> List[Dict[str, Any]]:
    delete_targets: List[Dict[str, Any]] = []
    seen_targets: Set[tuple[str, str]] = set()
    live_by_id = current_by_id or {}
    live_files = current_files or set()

    for row in stale_rows:
        entry_id = row.get("magento_entry_id")
        magento_file = str(row.get("magento_file") or "").strip()
        remote_url = row.get("remote_url")

        if entry_id is not None:
            live_entry = live_by_id.get(int(entry_id))
            file_name = str((live_entry or {}).get("file") or "") or magento_file
            key = ("id", str(int(entry_id)))
            if key not in seen_targets:
                seen_targets.add(key)
                delete_targets.append(
                    {
                        "entry_id": int(entry_id),
                        "file": file_name,
                        "remote_url": remote_url,
                        "reason": "stale_map",
                    }
                )
            continue

        if magento_file and magento_file in live_files:
            key = ("file", magento_file)
            if key not in seen_targets:
                seen_targets.add(key)
                delete_targets.append(
                    {
                        "entry_id": None,
                        "file": magento_file,
                        "remote_url": remote_url,
                        "reason": "stale_map",
                    }
                )

    return delete_targets


def _keep_entry_ids_for_desired(
    map_rows: Sequence[Dict[str, Any]],
    desired_set: Set[str],
) -> Set[int]:
    """Magento gallery entry ids that map to a current master URL."""
    keep: Set[int] = set()
    for row in map_rows:
        url = str(row.get("remote_url") or "").strip()
        if not url or url not in desired_set:
            continue
        entry_id = row.get("magento_entry_id")
        if entry_id is None:
            continue
        keep.add(int(entry_id))
    return keep


def _extend_unmapped_delete_targets(
    delete_targets: List[Dict[str, Any]],
    *,
    current_entries: Sequence[Dict[str, Any]],
    keep_entry_ids: Set[int],
) -> List[Dict[str, Any]]:
    """Also delete live Magento entries that are not mapped to any desired master URL."""
    seen_ids = {
        int(target["entry_id"])
        for target in delete_targets
        if target.get("entry_id") is not None
    }
    for entry in current_entries:
        if not isinstance(entry, dict) or entry.get("id") is None:
            continue
        entry_id = int(entry["id"])
        if entry_id in keep_entry_ids or entry_id in seen_ids:
            continue
        seen_ids.add(entry_id)
        delete_targets.append(
            {
                "entry_id": entry_id,
                "file": str(entry.get("file") or "").strip(),
                "remote_url": None,
                "reason": "unmapped",
            }
        )
    return delete_targets


def _entries_from_product_payload(product: Any) -> List[Dict[str, Any]]:
    if not isinstance(product, dict):
        return []
    raw = product.get("media_gallery_entries") or []
    if not isinstance(raw, list):
        return []
    return [entry for entry in raw if isinstance(entry, dict) and entry.get("id") is not None]


def list_product_media_entries(api: Any, sku: str) -> tuple[int, List[Dict[str, Any]], Optional[str], str]:
    """List gallery entries, falling back when Magento /media fails on missing files.

    Magento GET /V1/products/{sku}/media often returns HTTP 400 when any gallery file
    is missing on disk ("The contents from the file can't be read"). Product GET still
    returns ``media_gallery_entries`` with ids, which is enough for purge/delete.

    Returns ``(status, entries, error, source)`` where source is ``media_endpoint`` or
    ``product_payload``.
    """
    status, entries, err = api.get_product_media(sku)
    if status == 200:
        return status, list(entries or []), None, "media_endpoint"

    logger.warning(
        "get_product_media failed for %s status=%s error=%s; trying product.media_gallery_entries",
        sku,
        status,
        err,
    )
    product = None
    get_product = getattr(api, "get_product", None)
    if not callable(get_product):
        return status, [], err or f"HTTP {status}", "media_endpoint"
    try:
        product = get_product(sku)
    except Exception as exc:  # noqa: BLE001 - surface as listing failure
        return status, [], f"{err or f'HTTP {status}'}; get_product fallback failed: {exc}", "media_endpoint"

    if product is None and status in (400, 404):
        # Confirmed missing (get_product already tried slash path encodings + search).
        return 404, [], err or f"HTTP {status}", "product_not_found"

    fallback = _entries_from_product_payload(product)
    if fallback:
        return 200, fallback, err, "product_payload"
    if isinstance(product, dict):
        # Product exists but has an empty gallery — treat as successful empty list.
        return 200, [], None, "product_payload"
    return status, [], err or f"HTTP {status}", "media_endpoint"


def wipe_product_media(
    *,
    connection_id: int,
    sku: str,
    api: Any,
    media_repo: Any,
    catalog_repo: Any,
    dry_run: bool = False,
) -> Dict[str, Any]:
    """Delete every Magento gallery entry + media-map row for one SKU.

    Use when media_map keep-ids are untrusted (e.g. mapped to wrong on-disk files) and
    the gallery must be rebuilt from master via a subsequent force image push.
    """
    status, entries, err, media_source = list_product_media_entries(api, sku)
    if status != 200:
        return {
            "status": "failed",
            "sku": sku,
            "connection_id": connection_id,
            "error": err or f"media listing returned HTTP {status}",
            "upstream_status": status,
            "media_source": media_source,
            "delete_targets": [],
            "deleted_entries": [],
            "removed_map_rows": 0,
            "wipe_all": True,
        }

    current_entries = list(entries or [])
    delete_targets = [
        {
            "entry_id": int(entry["id"]),
            "file": str(entry.get("file") or "").strip(),
            "remote_url": None,
            "reason": "wipe_all",
        }
        for entry in current_entries
        if isinstance(entry, dict) and entry.get("id") is not None
    ]

    if dry_run:
        map_rows = list(media_repo.list_rows_for_sku(connection_id, sku))
        return {
            "status": "dry_run",
            "sku": sku,
            "connection_id": connection_id,
            "media_source": media_source,
            "wipe_all": True,
            "desired_urls": [],
            "keep_entry_ids": [],
            "delete_targets": delete_targets,
            "deleted_entries": [],
            "removed_map_rows": 0,
            "media_entry_count_before": len(current_entries),
            "media_entry_count_after": 0,
            "media_map_rows_before": len(map_rows),
            "media_files_after": current_entries,
        }

    deleted_entries: List[Dict[str, Any]] = []
    delete_errors: List[Dict[str, Any]] = []
    for target in delete_targets:
        entry_id = int(target["entry_id"])
        del_status, del_err = api.delete_product_media(sku, entry_id)
        if del_status in (200, 204):
            deleted_entries.append(target)
        else:
            delete_errors.append(
                {
                    "entry_id": entry_id,
                    "file": target.get("file"),
                    "error": del_err or f"HTTP {del_status}",
                }
            )

    removed_map_rows = 0
    if hasattr(media_repo, "delete_all_for_sku"):
        removed_map_rows = int(media_repo.delete_all_for_sku(connection_id, sku))

    status_after, entries_after, err_after, source_after = list_product_media_entries(api, sku)
    refreshed_entries = list(entries_after or []) if status_after == 200 else []
    refreshed_media_hash = _media_hash(refreshed_entries) if refreshed_entries else None
    if status_after == 200:
        catalog_repo.upsert_catalog_state(
            connection_id,
            sku,
            media_files=refreshed_entries,
            media_hash=refreshed_media_hash,
        )

    status_out = "ok" if not delete_errors and status_after == 200 and not refreshed_entries else "partial"
    if delete_errors:
        status_out = "partial"
    return {
        "status": status_out,
        "sku": sku,
        "connection_id": connection_id,
        "media_source": source_after if status_after == 200 else media_source,
        "wipe_all": True,
        "desired_urls": [],
        "keep_entry_ids": [],
        "delete_targets": delete_targets,
        "deleted_entries": deleted_entries,
        "removed_map_rows": removed_map_rows,
        "delete_errors": delete_errors,
        "media_entry_count_before": len(current_entries),
        "media_entry_count_after": len(refreshed_entries),
        "media_hash_after": refreshed_media_hash,
        "media_files_after": refreshed_entries,
        "error": None if status_after == 200 else (err_after or f"post-wipe listing HTTP {status_after}"),
        "upstream_status": status_after if status_after != 200 else None,
    }


def cleanup_orphaned_product_media(
    *,
    connection_id: int,
    sku: str,
    desired_urls: Sequence[str],
    api: Any,
    media_repo: Any,
    catalog_repo: Any,
    dry_run: bool = False,
    purge_unmapped: bool = False,
) -> Dict[str, Any]:
    """Delete Magento gallery entries that are no longer desired for one SKU.

    Default (conservative):
    - keep every master URL in `desired_urls`
    - only delete Magento entries linked to non-master media-map rows
    - remove those stale media-map rows from the DB
    - refresh magento_catalog_state.media_files/media_hash from live Magento

    With ``purge_unmapped=True``:
    - also delete live Magento gallery entries whose entry id is not mapped to
      any desired master URL (fixes stale caramel/wrong assets left by append-only sync)
    """
    normalized_desired: List[str] = []
    seen: Set[str] = set()
    for raw in desired_urls:
        url = str(raw or "").strip()
        if not url or url in seen:
            continue
        seen.add(url)
        normalized_desired.append(url)

    map_rows = list(media_repo.list_rows_for_sku(connection_id, sku))
    desired_set = set(normalized_desired)
    keep_entry_ids = _keep_entry_ids_for_desired(map_rows, desired_set)
    stale_rows = [row for row in map_rows if str(row.get("remote_url") or "").strip() not in desired_set]
    stale_remote_urls = [str(row.get("remote_url") or "").strip() for row in stale_rows if str(row.get("remote_url") or "").strip()]

    status, entries, err, media_source = list_product_media_entries(api, sku)
    if status != 200:
        logger.warning(
            "cleanup_orphaned_product_media media listing failed: connection_id=%s sku=%s status=%s source=%s error=%s",
            connection_id,
            sku,
            status,
            media_source,
            err,
        )
        if media_source == "product_not_found" or status == 404:
            removed_map_rows = 0
            if not dry_run and map_rows:
                # Product truly absent — drop stale local media-map rows only.
                stale_urls = [
                    str(row.get("remote_url") or "").strip()
                    for row in map_rows
                    if str(row.get("remote_url") or "").strip()
                ]
                if stale_urls:
                    removed_map_rows = int(media_repo.delete_for_sku_urls(connection_id, sku, stale_urls))
            return {
                "status": "skipped" if not dry_run else "dry_run",
                "error": err or "Product not found in Magento",
                "sku": sku,
                "connection_id": connection_id,
                "upstream_status": status,
                "media_source": media_source or "product_not_found",
                "desired_urls": normalized_desired,
                "stale_remote_urls": stale_remote_urls,
                "purge_unmapped": purge_unmapped,
                "keep_entry_ids": sorted(keep_entry_ids),
                "delete_targets": [],
                "deleted_entries": [],
                "removed_map_rows": removed_map_rows,
                "delete_errors": [],
            }

        fallback_targets = _build_delete_targets(stale_rows)
        if dry_run:
            return {
                "status": "dry_run",
                "error": err or f"get_product_media returned HTTP {status}",
                "sku": sku,
                "connection_id": connection_id,
                "upstream_status": status,
                "media_source": media_source,
                "desired_urls": normalized_desired,
                "stale_remote_urls": stale_remote_urls,
                "purge_unmapped": purge_unmapped,
                "keep_entry_ids": sorted(keep_entry_ids),
                "delete_targets": fallback_targets,
            }

        deleted_entries: List[Dict[str, Any]] = []
        delete_errors: List[Dict[str, Any]] = []
        for target in fallback_targets:
            entry_id = target.get("entry_id")
            if entry_id is None:
                delete_errors.append(
                    {
                        "remote_url": target.get("remote_url"),
                        "file": target.get("file"),
                        "error": "Fallback deletion requires magento_entry_id",
                    }
                )
                continue
            del_status, del_err = api.delete_product_media(sku, int(entry_id))
            if del_status in (200, 204):
                deleted_entries.append(target)
            else:
                delete_errors.append(
                    {
                        "entry_id": entry_id,
                        "remote_url": target.get("remote_url"),
                        "file": target.get("file"),
                        "error": del_err or f"HTTP {del_status}",
                    }
                )

        deleted_remote_urls = [str(item.get("remote_url") or "").strip() for item in deleted_entries if str(item.get("remote_url") or "").strip()]
        removed_map_rows = 0
        if deleted_remote_urls:
            removed_map_rows = int(media_repo.delete_for_sku_urls(connection_id, sku, deleted_remote_urls))

        status_after, entries_after, err_after, source_after = list_product_media_entries(api, sku)
        if status_after != 200:
            return {
                "status": "partial",
                "error": err_after or f"post-delete media listing returned HTTP {status_after}",
                "sku": sku,
                "connection_id": connection_id,
                "upstream_status": status_after,
                "media_source": source_after,
                "desired_urls": normalized_desired,
                "stale_remote_urls": stale_remote_urls,
                "purge_unmapped": purge_unmapped,
                "keep_entry_ids": sorted(keep_entry_ids),
                "delete_targets": fallback_targets,
                "deleted_entries": deleted_entries,
                "removed_map_rows": removed_map_rows,
                "delete_errors": delete_errors,
            }

        refreshed_entries = list(entries_after or [])
        refreshed_media_hash = _media_hash(refreshed_entries) if refreshed_entries else None
        catalog_repo.upsert_catalog_state(
            connection_id,
            sku,
            media_files=refreshed_entries,
            media_hash=refreshed_media_hash,
        )
        return {
            "status": "ok" if not delete_errors else "partial",
            "error": err or f"initial media listing returned HTTP {status}",
            "sku": sku,
            "connection_id": connection_id,
            "upstream_status": status,
            "media_source": media_source,
            "desired_urls": normalized_desired,
            "stale_remote_urls": stale_remote_urls,
            "purge_unmapped": purge_unmapped,
            "keep_entry_ids": sorted(keep_entry_ids),
            "delete_targets": fallback_targets,
            "deleted_entries": deleted_entries,
            "removed_map_rows": removed_map_rows,
            "delete_errors": delete_errors,
            "media_entry_count_after": len(refreshed_entries),
            "media_hash_after": refreshed_media_hash,
            "media_files_after": refreshed_entries,
        }

    current_entries = list(entries or [])
    current_by_id = {
        int(entry["id"]): entry
        for entry in current_entries
        if isinstance(entry, dict) and entry.get("id") is not None
    }
    current_files = {
        str(entry.get("file") or "").strip()
        for entry in current_entries
        if isinstance(entry, dict) and str(entry.get("file") or "").strip()
    }

    delete_targets = _build_delete_targets(
        stale_rows,
        current_by_id=current_by_id,
        current_files=current_files,
    )
    if purge_unmapped:
        delete_targets = _extend_unmapped_delete_targets(
            delete_targets,
            current_entries=current_entries,
            keep_entry_ids=keep_entry_ids,
        )

    deleted_entries: List[Dict[str, Any]] = []
    delete_errors: List[Dict[str, Any]] = []
    if not dry_run:
        for target in delete_targets:
            entry_id = target.get("entry_id")
            if entry_id is None:
                delete_errors.append(
                    {
                        "remote_url": target.get("remote_url"),
                        "file": target.get("file"),
                        "error": "Current Magento entry id not available for deletion",
                    }
                )
                continue
            del_status, del_err = api.delete_product_media(sku, int(entry_id))
            if del_status in (200, 204):
                deleted_entries.append(target)
            else:
                delete_errors.append(
                    {
                        "entry_id": entry_id,
                        "remote_url": target.get("remote_url"),
                        "file": target.get("file"),
                        "error": del_err or f"HTTP {del_status}",
                    }
                )

    deleted_remote_urls = [str(item.get("remote_url") or "").strip() for item in deleted_entries if str(item.get("remote_url") or "").strip()]
    removed_map_rows = 0
    if not dry_run and deleted_remote_urls:
        removed_map_rows = int(media_repo.delete_for_sku_urls(connection_id, sku, deleted_remote_urls))

    refreshed_entries = current_entries
    refreshed_media_hash = _media_hash(refreshed_entries) if refreshed_entries else None
    if not dry_run:
        status_after, entries_after, err_after, source_after = list_product_media_entries(api, sku)
        if status_after != 200:
            logger.warning(
                "cleanup_orphaned_product_media post-delete refresh failed: connection_id=%s sku=%s status=%s error=%s",
                connection_id,
                sku,
                status_after,
                err_after,
            )
            return {
                "status": "partial",
                "sku": sku,
                "connection_id": connection_id,
                "media_source": media_source,
                "desired_urls": normalized_desired,
                "stale_remote_urls": stale_remote_urls,
                "purge_unmapped": purge_unmapped,
                "keep_entry_ids": sorted(keep_entry_ids),
                "delete_targets": delete_targets,
                "deleted_entries": deleted_entries,
                "removed_map_rows": removed_map_rows,
                "delete_errors": delete_errors,
                "error": err_after or f"post-delete media listing returned HTTP {status_after}",
                "upstream_status": status_after,
            }
        refreshed_entries = list(entries_after or [])
        refreshed_media_hash = _media_hash(refreshed_entries) if refreshed_entries else None
        catalog_repo.upsert_catalog_state(
            connection_id,
            sku,
            media_files=refreshed_entries,
            media_hash=refreshed_media_hash,
        )
        media_source = source_after or media_source

    return {
        "status": "dry_run" if dry_run else ("ok" if not delete_errors else "partial"),
        "sku": sku,
        "connection_id": connection_id,
        "media_source": media_source,
        "desired_urls": normalized_desired,
        "stale_remote_urls": stale_remote_urls,
        "purge_unmapped": purge_unmapped,
        "keep_entry_ids": sorted(keep_entry_ids),
        "delete_targets": delete_targets,
        "deleted_entries": deleted_entries,
        "removed_map_rows": removed_map_rows,
        "delete_errors": delete_errors,
        "media_entry_count_before": len(current_entries),
        "media_entry_count_after": len(refreshed_entries) if not dry_run else len(current_entries) - len(
            [t for t in delete_targets if t.get("entry_id") is not None]
        ),
        "media_hash_after": refreshed_media_hash,
        "media_files_after": refreshed_entries,
    }


def desired_image_urls_for_sku(session: Any, sku: str) -> List[str]:
    """Master gallery URLs for one SKU (base + additional), stable-unique."""
    from db.channel_image_push import image_urls_from_fields
    from db.master_product_images import image_fields_by_sku

    fields = image_fields_by_sku(session, [sku]).get(sku) or {}
    return image_urls_from_fields(fields)


def resolve_orphan_cleanup_skus(
    session: Any,
    *,
    skus: Optional[Sequence[str]] = None,
    all_assigned: bool = False,
    offset: int = 0,
    limit: Optional[int] = None,
) -> List[str]:
    """Resolve Magento cleanup targets: explicit SKUs or all channel-assigned SKUs."""
    from db.channel_exports import resolve_push_skus

    if all_assigned:
        wanted = resolve_push_skus(session, "magento", skus=None, only_assigned=True)
    elif skus:
        wanted = resolve_push_skus(session, "magento", skus=list(skus), only_assigned=False)
    else:
        return []

    ordered = sorted({str(sku).strip() for sku in wanted if str(sku).strip()})
    start = max(0, int(offset or 0))
    if start:
        ordered = ordered[start:]
    if limit is not None:
        ordered = ordered[: max(0, int(limit))]
    return ordered


def cleanup_orphaned_media_batch(
    *,
    connection_id: int,
    skus: Sequence[str],
    api: Any,
    media_repo: Any,
    catalog_repo: Any,
    session: Any,
    dry_run: bool = False,
    purge_unmapped: bool = True,
    batch_size: int = DEFAULT_ORPHAN_CLEANUP_BATCH_SIZE,
    on_progress: Optional[ProgressCallback] = None,
    include_sku_results: bool = False,
) -> Dict[str, Any]:
    """Delete orphan Magento gallery entries for many SKUs in batches.

    Keeps master desired images; with ``purge_unmapped=True`` (default) also deletes
    live gallery entries that are not mapped to any current master URL.
    Does not wipe galleries and does not re-upload images.
    """
    clean_skus = [str(sku).strip() for sku in skus if str(sku or "").strip()]
    # stable unique
    seen: Set[str] = set()
    ordered: List[str] = []
    for sku in clean_skus:
        if sku in seen:
            continue
        seen.add(sku)
        ordered.append(sku)

    batch_size = max(1, int(batch_size or DEFAULT_ORPHAN_CLEANUP_BATCH_SIZE))
    total = len(ordered)
    processed = 0
    ok_count = 0
    skipped_count = 0
    partial_count = 0
    failed_count = 0
    deleted_entry_count = 0
    deleted_map_row_count = 0
    delete_target_count = 0
    sku_results: List[Dict[str, Any]] = []
    failure_samples: List[Dict[str, Any]] = []

    def _emit(stage: str, **extra: Any) -> None:
        if not on_progress:
            return
        payload = {
            "stage": stage,
            "total": total,
            "completed": processed,
            "success_count": ok_count + skipped_count,
            "failed": failed_count + partial_count,
            "notes": (
                f"orphan cleanup {processed}/{total} "
                f"(ok={ok_count} skipped={skipped_count} partial={partial_count} failed={failed_count})"
            ),
            **extra,
        }
        on_progress(payload)

    _emit("starting")

    for batch_start in range(0, total, batch_size):
        batch = ordered[batch_start : batch_start + batch_size]
        batch_index = batch_start // batch_size + 1
        batch_count = (total + batch_size - 1) // batch_size
        logger.info(
            "orphan cleanup batch %s/%s connection_id=%s size=%s dry_run=%s purge_unmapped=%s",
            batch_index,
            batch_count,
            connection_id,
            len(batch),
            dry_run,
            purge_unmapped,
        )

        for sku in batch:
            desired = desired_image_urls_for_sku(session, sku)
            if not desired:
                row = {
                    "sku": sku,
                    "status": "skipped",
                    "error": "No master images for SKU",
                    "desired_urls": [],
                    "purge_unmapped": purge_unmapped,
                }
                skipped_count += 1
                processed += 1
                if include_sku_results:
                    sku_results.append(row)
                _emit("sku", current_sku=sku, batch_index=batch_index, batch_count=batch_count)
                continue

            result = cleanup_orphaned_product_media(
                connection_id=connection_id,
                sku=sku,
                desired_urls=desired,
                api=api,
                media_repo=media_repo,
                catalog_repo=catalog_repo,
                dry_run=dry_run,
                purge_unmapped=purge_unmapped,
            )
            status = str(result.get("status") or "")
            if status in {"ok", "dry_run"}:
                ok_count += 1
            elif status == "skipped":
                skipped_count += 1
            elif status == "partial":
                partial_count += 1
            elif status == "failed":
                failed_count += 1
            else:
                # treat unknown as partial so operators can inspect samples
                partial_count += 1

            targets = result.get("delete_targets") or []
            deleted = result.get("deleted_entries") or []
            delete_target_count += len(targets)
            deleted_entry_count += len(deleted)
            deleted_map_row_count += int(result.get("removed_map_rows") or 0)

            compact = {
                "sku": sku,
                "status": status,
                "desired_url_count": len(desired),
                "delete_target_count": len(targets),
                "deleted_entry_count": len(deleted),
                "removed_map_rows": int(result.get("removed_map_rows") or 0),
                "error": result.get("error"),
                "upstream_status": result.get("upstream_status"),
            }
            if include_sku_results:
                sku_results.append(result if dry_run else compact)
            if status in {"failed", "partial"} and len(failure_samples) < 50:
                failure_samples.append(compact)

            processed += 1
            _emit("sku", current_sku=sku, batch_index=batch_index, batch_count=batch_count)

        if not dry_run:
            try:
                session.commit()
            except Exception:
                logger.exception(
                    "orphan cleanup commit failed after batch %s/%s connection_id=%s",
                    batch_index,
                    batch_count,
                    connection_id,
                )
                session.rollback()
                raise

        _emit("batch_done", batch_index=batch_index, batch_count=batch_count)

    overall = "ok"
    if failed_count and (ok_count or skipped_count or partial_count):
        overall = "partial"
    elif failed_count and not (ok_count or skipped_count or partial_count):
        overall = "failed"
    elif partial_count and not failed_count:
        overall = "partial"
    elif dry_run:
        overall = "dry_run"

    summary = {
        "status": overall,
        "connection_id": connection_id,
        "dry_run": dry_run,
        "purge_unmapped": purge_unmapped,
        "mode": "orphans_only",
        "batch_size": batch_size,
        "sku_count": total,
        "processed": processed,
        "ok_count": ok_count,
        "skipped_count": skipped_count,
        "partial_count": partial_count,
        "failed_count": failed_count,
        "delete_target_count": delete_target_count,
        "deleted_entry_count": deleted_entry_count,
        "deleted_map_row_count": deleted_map_row_count,
        "failure_samples": failure_samples,
        "total_count": total,
        "success_count": ok_count + skipped_count,
        "error_count": failed_count + partial_count,
        "message": (
            f"Orphan media cleanup {'dry-run ' if dry_run else ''}"
            f"processed {processed}/{total} SKUs "
            f"(deleted_entries={deleted_entry_count}, map_rows={deleted_map_row_count})."
        ),
    }
    if include_sku_results:
        summary["results"] = sku_results
    _emit("done")
    return summary