from __future__ import annotations
import asyncio
import html as _html_module
import logging
import re
import shutil
import time
from decimal import Decimal
from pathlib import Path

import httpx
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger("uvicorn.error")

from app.core.models.brand import Brand
from app.core.models.category import Category
from app.core.models.product import Product, ProductImage, ProductVariant, ProductTierPrice, product_categories

def _unescape_desc(text: str | None) -> str | None:
    """
    OpenCart 部分安装会把描述以 HTML 实体形式存储（&lt;p&gt; 而非 <p>）。
    导入时统一反转义，保证 Store v-html 能正确渲染。
    """
    if not text:
        return text
    if "&lt;" in text or "&amp;" in text or "&gt;" in text:
        return _html_module.unescape(text)
    return text


try:
    from slugify import slugify as _slugify_lib

    def make_slug(text: str, fallback: str = "") -> str:
        result = _slugify_lib(text, allow_unicode=True, separator="-")
        return result or fallback
except ImportError:
    def make_slug(text: str, fallback: str = "") -> str:
        slug = re.sub(r"[^\w\s-]", "", text.lower())
        slug = re.sub(r"[-\s]+", "-", slug).strip("-")
        return slug or fallback


async def _download_image(
    client: httpx.AsyncClient,
    src_url: str,
    dst: Path,
) -> bool:
    """从旧站下载一张图片到本地，返回是否成功。"""
    if dst.exists():
        return True  # 已存在，跳过
    try:
        resp = await client.get(src_url, timeout=15, follow_redirects=True)
        if resp.status_code == 200 and resp.content:
            dst.parent.mkdir(parents=True, exist_ok=True)
            dst.write_bytes(resp.content)
            return True
        logger.warning("OC image download %s → HTTP %s", src_url, resp.status_code)
        return False
    except Exception as exc:
        logger.warning("OC image download failed %s: %s", src_url, exc)
        return False


async def _download_images_batch(
    img_paths: list[str],
    oc_img_base: str,
    static_uploads_dir: Path,
    concurrency: int = 8,
) -> tuple[int, int]:
    """
    批量并发下载图片。
    返回 (downloaded, failed)。
    img_paths: OpenCart 原始路径列表，如 "catalog/product/xxx.jpg"
    oc_img_base: 旧站根地址，如 "https://www.oldstore.com"
    """
    downloaded = failed = 0
    sem = asyncio.Semaphore(concurrency)

    async def fetch_one(img_path: str):
        nonlocal downloaded, failed
        src_url = f"{oc_img_base}/image/{img_path.lstrip('/')}"
        dst = static_uploads_dir / img_path.lstrip("/")
        async with sem:
            ok = await _download_image(client, src_url, dst)
        if ok:
            downloaded += 1
        else:
            failed += 1

    async with httpx.AsyncClient(
        headers={"User-Agent": "LS/1.0 ImageMigrator"},
        limits=httpx.Limits(max_connections=concurrency),
    ) as client:
        await asyncio.gather(*[fetch_one(p) for p in img_paths])

    return downloaded, failed


def make_unique_slug(base: str, existing: set[str]) -> str:
    if base not in existing:
        return base
    i = 2
    while f"{base}-{i}" in existing:
        i += 1
    return f"{base}-{i}"


# ── brands ────────────────────────────────────────────────────

_IMG_CT: dict[str, str] = {
    "jpg": "image/jpeg", "jpeg": "image/jpeg",
    "png": "image/png", "webp": "image/webp",
    "gif": "image/gif",
}


def _resolve_img_url(img_path: str, image_base_url: str, oc_img_base: str = "") -> str:
    """将图片路径解析为完整 URL。
    - 已经是完整 URL（导出脚本内嵌了旧站地址）→ 直接返回
    - oc_img_base 不为空 → 拼接旧站地址
    - 否则 → 拼接本地静态地址
    """
    if img_path.startswith(("http://", "https://")):
        return img_path
    if oc_img_base:
        return f"{oc_img_base}/image/{img_path.lstrip('/')}"
    return image_base_url.rstrip("/") + "/" + img_path.lstrip("/")

def _dict_or_empty(value) -> dict:
    return value if isinstance(value, dict) else {}


def _lang_desc(values, language_id: int | None, fallback: bool = False) -> dict:
    values = _dict_or_empty(values)
    if language_id is not None:
        desc = _dict_or_empty(values.get(str(language_id)))
        if desc:
            return desc
    return _dict_or_empty(next(iter(values.values()), {})) if fallback and values else {}

def map_brand(oc_brand: dict, tenant_id: int, image_base_url: str, oc_img_base: str = "") -> dict:
    logo_url = None
    if oc_brand.get("logo_path"):
        logo_url = _resolve_img_url(oc_brand["logo_path"], image_base_url, oc_img_base)
    return {
        "tenant_id":  tenant_id,
        "name":       oc_brand["name"],
        "slug":       make_slug(oc_brand["name"], fallback=f"brand-{oc_brand['id']}"),
        "logo_url":   logo_url,
        "sort_order": oc_brand.get("sort_order", 0),
        "is_active":  1,
    }


async def import_brands(
    db: AsyncSession,
    oc_brands: list[dict],
    tenant_id: int,
    image_base_url: str,
    conflict_strategy: str,
    oc_img_base: str = "",
) -> tuple[int, int]:
    """Returns (created, skipped)."""
    created = skipped = 0
    existing_slugs: set[str] = set()

    result = await db.execute(select(Brand.slug).where(Brand.tenant_id == tenant_id))
    for row in result.scalars():
        existing_slugs.add(row)

    for oc_b in oc_brands:
        existing_result = await db.execute(
            select(Brand).where(Brand.tenant_id == tenant_id, Brand.name == oc_b["name"])
        )
        existing_brand = existing_result.scalars().first()

        if existing_brand:
            if conflict_strategy == "overwrite":
                data = map_brand(oc_b, tenant_id, image_base_url, oc_img_base)
                for k, v in data.items():
                    if k not in ("slug", "tenant_id"):
                        setattr(existing_brand, k, v)
            skipped += 1
        else:
            data = map_brand(oc_b, tenant_id, image_base_url, oc_img_base)
            slug = make_unique_slug(data["slug"], existing_slugs)
            data["slug"] = slug
            existing_slugs.add(slug)
            db.add(Brand(**data))
            created += 1

    await db.flush()
    return created, skipped


# ── categories ───────────────────────────────────────────────

def map_category(
    oc_cat: dict,
    language_id: int,
    tenant_id: int,
    cat_id_map: dict[int, int],
    image_base_url: str,
    oc_img_base: str = "",
    english_language_id: int | None = None,
) -> dict:
    names = _dict_or_empty(oc_cat.get("names"))
    desc = _lang_desc(names, language_id, fallback=True)
    en_desc = _lang_desc(names, english_language_id)
    name = desc.get("name", "") or f"category-{oc_cat['id']}"
    image_url = None
    if oc_cat.get("image_path"):
        image_url = _resolve_img_url(oc_cat["image_path"], image_base_url, oc_img_base)

    parent_oc_id = oc_cat.get("parent_id", 0)
    parent_id = cat_id_map.get(parent_oc_id) if parent_oc_id else None

    return {
        "tenant_id":        tenant_id,
        "name":             name,
        "slug":             make_slug(name, fallback=f"cat-{oc_cat['id']}"),
        "description":      _unescape_desc(desc.get("description")) or None,
        "meta_title":       desc.get("meta_title") or None,
        "meta_description": desc.get("meta_description") or None,
        "seo_keywords":     desc.get("meta_keyword") or None,
        "name_en":          en_desc.get("name") or None,
        "description_en":   _unescape_desc(en_desc.get("description")) or None,
        "meta_title_en":    en_desc.get("meta_title") or None,
        "meta_description_en": en_desc.get("meta_description") or None,
        "seo_keywords_en":  en_desc.get("meta_keyword") or None,
        "image_url":        image_url,
        "parent_id":        parent_id,
        "sort_order":       oc_cat.get("sort_order", 0),
        "is_active":        1 if oc_cat.get("status", 1) else 0,
    }


async def import_categories(
    db: AsyncSession,
    oc_categories: list[dict],
    language_id: int,
    tenant_id: int,
    image_base_url: str,
    conflict_strategy: str,
    oc_img_base: str = "",
    english_language_id: int | None = None,
) -> tuple[int, dict[int, int]]:
    """Returns (created_count, oc_id_to_new_id mapping)."""
    created = 0
    cat_id_map: dict[int, int] = {}
    existing_slugs: set[str] = set()

    result = await db.execute(select(Category.slug).where(Category.tenant_id == tenant_id))
    for row in result.scalars():
        existing_slugs.add(row)

    # top-level first, then children
    sorted_cats = sorted(oc_categories, key=lambda c: (0 if c["parent_id"] == 0 else 1, c.get("sort_order", 0)))

    for oc_c in sorted_cats:
        data = map_category(oc_c, language_id, tenant_id, cat_id_map, image_base_url, oc_img_base, english_language_id)

        existing_result = await db.execute(
            select(Category).where(Category.tenant_id == tenant_id, Category.name == data["name"])
        )
        existing_cat = existing_result.scalars().first()

        if existing_cat:
            cat_id_map[oc_c["id"]] = existing_cat.id
            if conflict_strategy == "overwrite":
                for k, v in data.items():
                    if k not in ("slug", "tenant_id"):
                        setattr(existing_cat, k, v)
        else:
            slug = make_unique_slug(data["slug"], existing_slugs)
            data["slug"] = slug
            existing_slugs.add(slug)
            cat = Category(**data)
            db.add(cat)
            await db.flush()
            cat_id_map[oc_c["id"]] = cat.id
            created += 1

    return created, cat_id_map


# ── products ──────────────────────────────────────────────────

def resolve_price(oc_prod: dict, price_source: str) -> Decimal:
    if price_source == "product_price":
        return Decimal(str(oc_prod.get("price", "0") or "0"))
    if price_source.startswith("group_"):
        gid = int(price_source.split("_", 1)[1])
        for s in oc_prod.get("specials", []):
            if s["customer_group_id"] == gid and Decimal(str(s["price"] or "0")) > 0:
                return Decimal(str(s["price"]))
    return Decimal(str(oc_prod.get("price", "0") or "0"))


def map_product(
    oc_prod: dict,
    cfg: "MappingConfig",
    tenant_id: int,
    brand_id_map: dict[int, int],
    image_base_url: str,
    oc_img_base: str = "",
) -> dict:
    from app.plugins.opencart_import.schemas import MappingConfig  # noqa: F401
    descriptions = _dict_or_empty(oc_prod.get("descriptions"))
    desc = _lang_desc(descriptions, cfg.language_id, fallback=True)
    en_desc = _lang_desc(descriptions, getattr(cfg, "english_language_id", None))
    name = desc.get("name", "") or oc_prod["model"]

    cover_url = None
    if oc_prod.get("main_image"):
        cover_url = _resolve_img_url(oc_prod["main_image"], image_base_url, oc_img_base)

    def _nonzero_decimal(val: str, max_value: Decimal) -> Decimal | None:
        try:
            d = Decimal(str(val or "0"))
            return d if 0 < d <= max_value else None
        except Exception:
            return None

    return {
        "tenant_id":        tenant_id,
        "name":             name,
        "slug":             make_slug(name, fallback=oc_prod["model"]),
        "sku":              oc_prod["model"],
        "description":      _unescape_desc(desc.get("description")) or None,
        "base_price":       resolve_price(oc_prod, cfg.price_source),
        "stock_qty":        oc_prod.get("quantity", 0),
        "weight":           _nonzero_decimal(oc_prod.get("weight"), Decimal("99999.999")),
        "length":           _nonzero_decimal(oc_prod.get("length"), Decimal("999999.99")),
        "width":            _nonzero_decimal(oc_prod.get("width"), Decimal("999999.99")),
        "height":           _nonzero_decimal(oc_prod.get("height"), Decimal("999999.99")),
        "status":           "active" if oc_prod.get("status") == 1 else "draft",
        "brand_id":         brand_id_map.get(oc_prod.get("manufacturer_id", 0)),
        "cover_url":        cover_url,
        "meta_title":       desc.get("meta_title") or None,
        "seo_keywords":     desc.get("meta_keyword") or None,
        "meta_description": desc.get("meta_description") or None,
        "name_en":          en_desc.get("name") or None,
        "description_en":   _unescape_desc(en_desc.get("description")) or None,
        "meta_title_en":    en_desc.get("meta_title") or None,
        "meta_description_en": en_desc.get("meta_description") or None,
        "seo_keywords_en":  en_desc.get("meta_keyword") or None,
    }


async def import_products_stream(
    db: AsyncSession,
    oc_products: list[dict],
    cfg: "MappingConfig",
    tenant_id: int,
    brand_id_map: dict[int, int],
    cat_id_map: dict[int, int],
    image_base_url: str,
    tmp_images_dir: Path,
    static_uploads_dir: Path,
):
    """
    Async generator — yields SSE-style event dicts:
      {"type": "start",    "total": N}
      {"type": "progress", "done": i, "total": N, "created": K, "skipped": M, "overwritten": P}
      {"type": "error",    "sku": "...", "name": "...", "reason": "..."}   ← stops here
      {"type": "done",     "stats": {...}}
    遇到第一个失败立即 yield error 并 return，调用方负责 rollback。
    """
    from app.plugins.opencart_import.schemas import RunResponse, ImportError as OcImportError

    stats: dict = dict(
        brands_created=0, brands_skipped=0,
        categories_created=0,
        products_created=0, products_skipped=0, products_overwritten=0,
        variants_created=0, tier_prices_created=0,
        images_copied=0, images_missing=0,
        errors=[],
        skipped_skus=[],
    )

    total = len(oc_products)
    t_start = time.monotonic()
    logger.info("OC import START: %d products, tenant=%d, strategy=%s", total, tenant_id, cfg.conflict_strategy)

    yield {"type": "start", "total": total}

    # 预加载已有 slugs
    result = await db.execute(select(Product.slug).where(Product.tenant_id == tenant_id))
    existing_product_slugs: set[str] = {row for row in result.scalars()}
    logger.info("OC import: loaded %d existing slugs", len(existing_product_slugs))

    option_key_map: dict[int, str] = {m.option_id: m.attribute_key for m in cfg.option_mappings}
    cg_to_level: dict[int, int | None] = {m.customer_group_id: m.member_level_id for m in cfg.customer_group_mappings}

    oc_img_base = (getattr(cfg, "oc_image_base_url", "") or "").rstrip("/")
    # 填写了旧站域名时，跳过图片下载——图片 URL 暂时指向旧站，
    # 导入完成后用"迁移图片到CDN"步骤统一拉取并上传到配置的存储后端。
    if oc_img_base:
        logger.info("OC import: oc_img_base=%s — images will keep old-site URLs, sync to CDN post-import", oc_img_base)

    for i, oc_prod in enumerate(oc_products):
        sku = oc_prod["model"]
        desc = _dict_or_empty(oc_prod.get("descriptions"))
        prod_desc = _lang_desc(desc, cfg.language_id, fallback=True)
        prod_name = prod_desc.get("name", "")

        # 查重
        existing_result = await db.execute(
            select(Product).where(Product.tenant_id == tenant_id, Product.sku == sku)
        )
        existing_product = existing_result.scalars().first()

        if existing_product and cfg.conflict_strategy == "skip":
            stats["products_skipped"] += 1
            stats["skipped_skus"].append(sku)
        else:
            try:
                data = map_product(oc_prod, cfg, tenant_id, brand_id_map, image_base_url, oc_img_base)

                if existing_product:
                    for k, v in data.items():
                        if k not in ("slug", "tenant_id", "sku"):
                            setattr(existing_product, k, v)
                    product = existing_product
                    stats["products_overwritten"] += 1
                    await db.execute(ProductImage.__table__.delete().where(ProductImage.product_id == product.id))
                    await db.execute(ProductVariant.__table__.delete().where(ProductVariant.product_id == product.id))
                    await db.execute(ProductTierPrice.__table__.delete().where(ProductTierPrice.product_id == product.id))
                    await db.execute(product_categories.delete().where(product_categories.c.product_id == product.id))
                else:
                    await db.execute(
                        ProductVariant.__table__.delete().where(
                            ProductVariant.tenant_id == tenant_id,
                            ProductVariant.sku.like(f"{sku}_%"),
                        )
                    )
                    slug = make_unique_slug(data["slug"], existing_product_slugs)
                    data["slug"] = slug
                    existing_product_slugs.add(slug)
                    product = Product(**data)
                    db.add(product)
                    await db.flush()
                    stats["products_created"] += 1

                # 分类（去重）
                seen_cat_ids: set[int] = set()
                for oc_cat_id in oc_prod.get("category_ids", []):
                    new_cat_id = cat_id_map.get(oc_cat_id)
                    if new_cat_id and new_cat_id not in seen_cat_ids:
                        seen_cat_ids.add(new_cat_id)
                        await db.execute(
                            product_categories.insert().values(product_id=product.id, category_id=new_cat_id)
                        )

                # 图片
                sort_order = 0
                seen_imgs: set[str] = set()
                for img_path in ([oc_prod["main_image"]] if oc_prod.get("main_image") else []) + oc_prod.get("extra_images", []):
                    if not img_path or img_path in seen_imgs:
                        continue
                    seen_imgs.add(img_path)
                    if oc_img_base or img_path.startswith(("http://", "https://")):
                        # 完整 URL（导出脚本内嵌）或旧站域名模式：直接使用，后续可迁移到 CDN
                        full_url = _resolve_img_url(img_path, image_base_url, oc_img_base)
                    else:
                        src = tmp_images_dir / img_path.lstrip("/")
                        dst = static_uploads_dir / img_path.lstrip("/")
                        dst.parent.mkdir(parents=True, exist_ok=True)
                        if src.exists():
                            shutil.copy2(str(src), str(dst))
                            stats["images_copied"] += 1
                        else:
                            stats["images_missing"] += 1
                        full_url = image_base_url.rstrip("/") + "/" + img_path.lstrip("/")
                    db.add(ProductImage(product_id=product.id, tenant_id=tenant_id,
                                        url=full_url, sort_order=sort_order,
                                        is_primary=1 if sort_order == 0 else 0))
                    sort_order += 1

                # 规格变体
                for oc_opt in oc_prod.get("options", []):
                    attr_key = option_key_map.get(oc_opt["option_id"])
                    if not attr_key:
                        names = _dict_or_empty(oc_opt.get("option_names", {}))
                        attr_key = (names.get(str(cfg.language_id)) or next(iter(names.values()), None)) if names else None
                        attr_key = attr_key or f"option_{oc_opt['option_id']}"
                    for val in oc_opt.get("values", []):
                        val_names = _dict_or_empty(val.get("names", {}))
                        attr_val = val_names.get(str(cfg.language_id)) or next(iter(val_names.values()), str(val["option_value_id"]))
                        modifier = Decimal(str(val.get("price_modifier", "0") or "0"))
                        if val.get("price_prefix") == "-":
                            modifier = -modifier
                        variant_attrs = {attr_key: attr_val}
                        pack_qty = val.get("pack_qty", 1)
                        if pack_qty and pack_qty != 1:
                            variant_attrs["_pack_qty"] = pack_qty
                        variant = ProductVariant(
                            product_id=product.id, tenant_id=tenant_id,
                            sku=f"{sku}_{val['option_value_id']}",
                            price_modifier=modifier, stock_qty=oc_prod.get("quantity", 0),
                            attributes=variant_attrs, is_active=1, is_default=0,
                        )
                        async with db.begin_nested() as vsp:
                            try:
                                db.add(variant)
                                await db.flush()
                                stats["variants_created"] += 1
                            except IntegrityError:
                                await vsp.rollback()

                # 会员价
                for special in oc_prod.get("specials", []):
                    gid = special["customer_group_id"]
                    price = Decimal(str(special.get("price", "0") or "0"))
                    if price <= 0:
                        continue
                    level_id = cg_to_level.get(gid)
                    if level_id is None:
                        continue
                    db.add(ProductTierPrice(tenant_id=tenant_id, product_id=product.id,
                                             variant_id=None, member_level_id=level_id, price=price))
                    stats["tier_prices_created"] += 1

            except Exception as exc:
                logger.error("OC import failed sku=%s name=%s: %s", sku, prod_name, exc)
                yield {"type": "error", "sku": sku, "name": prod_name, "reason": str(exc)}
                return  # 立即停止，调用方负责 rollback

        # 每 50 个或最后一个推进度
        if (i + 1) % 50 == 0 or i == total - 1:
            elapsed = time.monotonic() - t_start
            logger.info("OC import progress: %d/%d in %.1fs (created=%d skipped=%d overwritten=%d)",
                        i + 1, total, elapsed,
                        stats["products_created"], stats["products_skipped"], stats["products_overwritten"])
            yield {
                "type": "progress",
                "done": i + 1,
                "total": total,
                "created": stats["products_created"],
                "skipped": stats["products_skipped"],
                "overwritten": stats["products_overwritten"],
                "elapsed": round(elapsed, 1),
            }

    logger.info("OC import DONE in %.1fs", time.monotonic() - t_start)
    yield {"type": "done", "stats": stats}


# ── post-import: sync images to CDN storage ───────────────────

async def sync_images_to_storage(
    db: AsyncSession,
    tenant_id: int,
    oc_img_base: str,
    storage,
    storage_prefix: str,
    concurrency: int = 6,
):
    """
    查找 DB 中所有指向旧 OpenCart 站的图片 URL，并发下载后直传到
    租户配置的存储后端（OSS / S3 / R2 / 本地），同步更新 DB 中的 URL。
    Yields SSE-style event dicts：
      {"type": "start", "total": N}
      {"type": "progress", "done": i, "total": N, "synced": K, "failed": M}
      {"type": "done", "synced": K, "failed": M}
    """
    from sqlalchemy import text as _text

    base = oc_img_base.rstrip("/")
    # 旧版导入器把品牌/分类图片存成了本地静态路径，需要一并处理
    _LOCAL_OC_PREFIX = "/api/static/uploads/opencart/"

    def _to_download_url(url: str) -> tuple[str, str]:
        """返回 (download_url, img_rel_path)。
        - 旧站完整 URL：直接下载，提取 /image/ 后的路径
        - 本地静态路径：用 oc_img_base 还原旧站 URL
        """
        if url.startswith(_LOCAL_OC_PREFIX):
            img_rel = url[len(_LOCAL_OC_PREFIX):]
            return f"{base}/image/{img_rel}", img_rel
        if "/image/" in url:
            img_rel = url.split("/image/", 1)[1]
        else:
            img_rel = url.split("/")[-1]
        return url, img_rel

    # 收集所有需要同步的 (table_kind, row_id, old_url)
    # 同时匹配：① 旧站 URL（base%）② 旧版导入留下的本地静态路径
    entries: list[tuple[str, int, str]] = []
    col_map = {
        "product_image": (
            "SELECT pi.id, pi.url FROM product_images pi "
            "JOIN products p ON pi.product_id = p.id "
            "WHERE p.tenant_id = :tid AND (pi.url LIKE :pat OR pi.url LIKE :local)",
        ),
        "product_cover": (
            "SELECT id, cover_url FROM products "
            "WHERE tenant_id = :tid AND (cover_url LIKE :pat OR cover_url LIKE :local)",
        ),
        "category": (
            "SELECT id, image_url FROM categories "
            "WHERE tenant_id = :tid AND (image_url LIKE :pat OR image_url LIKE :local)",
        ),
        "brand": (
            "SELECT id, logo_url FROM brands "
            "WHERE tenant_id = :tid AND (logo_url LIKE :pat OR logo_url LIKE :local)",
        ),
    }
    params = {"tid": tenant_id, "pat": f"{base}%", "local": f"{_LOCAL_OC_PREFIX}%"}
    for kind, (sql,) in col_map.items():
        rows = (await db.execute(_text(sql), params)).fetchall()
        for row in rows:
            entries.append((kind, row[0], row[1]))

    total = len(entries)
    yield {"type": "start", "total": total}

    if total == 0:
        yield {"type": "done", "synced": 0, "failed": 0}
        return

    synced = failed = 0
    url_cache: dict[str, str] = {}  # old_url -> cdn_url（同一张图片只上传一次）
    sem = asyncio.Semaphore(concurrency)

    async with httpx.AsyncClient(
        headers={"User-Agent": "LS/1.0 ImageMigrator"},
        limits=httpx.Limits(max_connections=concurrency),
        timeout=httpx.Timeout(30.0),
    ) as client:
        for i, (kind, row_id, old_url) in enumerate(entries):
            try:
                if old_url not in url_cache:
                    download_url, img_rel = _to_download_url(old_url)
                    storage_key = f"{storage_prefix}opencart/{img_rel}"
                    ext = img_rel.rsplit(".", 1)[-1].lower() if "." in img_rel else ""
                    content_type = _IMG_CT.get(ext, "image/jpeg")

                    async with sem:
                        resp = await client.get(download_url, follow_redirects=True)

                    if resp.status_code != 200 or not resp.content:
                        failed += 1
                        logger.warning("OC img sync HTTP %s: %s", resp.status_code, download_url)
                        if (i + 1) % 20 == 0 or i == total - 1:
                            yield {"type": "progress", "done": i + 1, "total": total,
                                   "synced": synced, "failed": failed}
                        continue

                    # 直传到 CDN/OSS/S3
                    new_url = await storage.save(storage_key, resp.content, content_type)
                    url_cache[old_url] = new_url
                    synced += 1

                new_url = url_cache[old_url]

                # 更新对应 DB 字段
                if kind == "product_image":
                    await db.execute(_text(
                        "UPDATE product_images SET url = :new WHERE id = :id"
                    ), {"new": new_url, "id": row_id})
                elif kind == "product_cover":
                    await db.execute(_text(
                        "UPDATE products SET cover_url = :new WHERE id = :id"
                    ), {"new": new_url, "id": row_id})
                elif kind == "category":
                    await db.execute(_text(
                        "UPDATE categories SET image_url = :new WHERE id = :id"
                    ), {"new": new_url, "id": row_id})
                elif kind == "brand":
                    await db.execute(_text(
                        "UPDATE brands SET logo_url = :new WHERE id = :id"
                    ), {"new": new_url, "id": row_id})

            except Exception as exc:
                failed += 1
                logger.warning("OC img sync failed %s: %s", old_url, exc)

            if (i + 1) % 20 == 0 or i == total - 1:
                yield {"type": "progress", "done": i + 1, "total": total,
                       "synced": synced, "failed": failed}
                await asyncio.sleep(0)

    await db.commit()
    yield {"type": "done", "synced": synced, "failed": failed}