"""回导：安全解包 → 解析（重拼分片）→ 跨文件分组 → 只读预检 → 复用规范写入路径提交。

写入语义（§9.4）：
- 商品创建/更新复用 products.py:create_product / update_product，不另写 ORM upsert。
- 更新前合并商品完整现有规格，避免 _apply_variants 删除文件未含的规格；合并的规格
  连同现有 tier_prices 等字段一并回填。
- 图片直写 ProductImage（现有 handler 自带 commit，不适合批内复用），强制租户校验、
  按 ID→URL 幂等匹配、缺行不删、主图互斥并同步 cover_url。
"""
from __future__ import annotations

import io
import json
import zipfile
from decimal import Decimal, InvalidOperation

from openpyxl import load_workbook
from pydantic import ValidationError
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.models.product import Product, ProductImage, ProductVariant, ProductTierPrice
from app.core.models.category import Category
from app.core.models.brand import Brand
from app.schemas.product import ProductCreate, ProductUpdate, ProductVariantIn, TierPriceIn
from .exporter import FORMAT_VERSION, COLUMNS, JSON_COLUMNS, READONLY_COLUMNS

# 压缩炸弹防护上限
MAX_FILES = 500
MAX_TOTAL_UNCOMPRESSED = 500 * 1024 * 1024   # 解压后总量 500MB
MAX_SINGLE_UNCOMPRESSED = 100 * 1024 * 1024  # 单文件 100MB


class ImportError_(Exception):
    """预检可解释错误：附带结构化明细。"""
    def __init__(self, errors: list[str]):
        self.errors = errors
        super().__init__("; ".join(errors[:5]))


# ── 安全解包 ────────────────────────────────────────────────────────
def safe_extract(data: bytes, filename: str) -> list[tuple[str, bytes]]:
    """返回 [(name, xlsx_bytes)]。单 XLSX 直接返回；ZIP 做炸弹/路径防护后解出。"""
    lower = (filename or "").lower()
    if lower.endswith(".xlsx"):
        if len(data) > MAX_SINGLE_UNCOMPRESSED:
            raise ImportError_(["XLSX 文件过大"])
        return [(filename, data)]
    if not lower.endswith(".zip"):
        raise ImportError_(["仅支持插件生成的 .xlsx 或 .zip"])

    out: list[tuple[str, bytes]] = []
    total = 0
    with zipfile.ZipFile(io.BytesIO(data)) as zf:
        infos = zf.infolist()
        if len(infos) > MAX_FILES:
            raise ImportError_([f"ZIP 内文件过多（>{MAX_FILES}）"])
        for info in infos:
            name = info.filename
            if name.endswith("/"):
                continue  # 目录项
            # 路径穿越 / 绝对路径 / 符号链接防护
            if name.startswith("/") or ".." in name.replace("\\", "/").split("/") or ":" in name:
                raise ImportError_([f"非法路径：{name}"])
            if (info.external_attr >> 16) & 0o170000 == 0o120000:  # S_IFLNK
                raise ImportError_([f"拒绝符号链接：{name}"])
            if not name.lower().endswith(".xlsx"):
                raise ImportError_([f"ZIP 内含非 XLSX 文件：{name}"])
            if info.file_size > MAX_SINGLE_UNCOMPRESSED:
                raise ImportError_([f"单文件解压过大：{name}"])
            total += info.file_size
            if total > MAX_TOTAL_UNCOMPRESSED:
                raise ImportError_(["解压后总量超限（疑似压缩炸弹）"])
            out.append((name, zf.read(info)))
    if not out:
        raise ImportError_(["ZIP 内无有效 XLSX"])
    return out


# ── 解析（重拼分片）─────────────────────────────────────────────────
def _rejoin_headers(headers: list[str]) -> dict[str, list[int]]:
    """base 列名 → 该列所有分片的列下标（按分片序）。"""
    order: dict[str, dict[int, int]] = {}
    for idx, h in enumerate(headers):
        if h is None:
            continue
        if "_part_" in h:
            base, num = h.rsplit("_part_", 1)
            try:
                part = int(num)
            except ValueError:
                base, part = h, 1
        else:
            base, part = h, 1
        order.setdefault(base, {})[part] = idx
    return {base: [order[base][k] for k in sorted(order[base])] for base in order}


def parse_xlsx(data: bytes, name: str) -> list[dict]:
    """XLSX → 行 dict 列表（分片已重拼）。校验表头与格式版本。"""
    wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True)
    ws = wb.active
    it = ws.iter_rows(values_only=True)
    try:
        headers = [str(h) if h is not None else None for h in next(it)]
    except StopIteration:
        raise ImportError_([f"{name}: 空文件"])
    col_map = _rejoin_headers(headers)
    if "format_version" not in col_map:
        raise ImportError_([f"{name}: 缺少 format_version 列"])
    missing = [c for c in COLUMNS if c not in col_map]
    if missing:
        raise ImportError_([f"{name}: 表头缺列 {missing[:8]}"])

    rows = []
    for raw in it:
        if raw is None or all(v is None or v == "" for v in raw):
            continue
        row = {}
        for base, idxs in col_map.items():
            parts = []
            for i in idxs:
                v = raw[i] if i < len(raw) else None
                parts.append("" if v is None else str(v))
            row[base] = "".join(parts)
        fv = row.get("format_version", "")
        if fv != FORMAT_VERSION:
            raise ImportError_([f"{name}: 不支持的格式版本 {fv!r}"])
        rows.append(row)
    return rows


def read_input(data: bytes, filename: str) -> list[dict]:
    """解包 + 解析所有文件为统一行列表；文件顺序不影响结果（后续按键分组）。"""
    all_rows = []
    for name, xbytes in safe_extract(data, filename):
        all_rows.extend(parse_xlsx(xbytes, name))
    return all_rows


# ── 分组 ────────────────────────────────────────────────────────────
# 商品公共字段（同一商品各行必须一致）
_PRODUCT_COMMON = [
    "product_sku", "name", "name_en", "slug", "category_path", "brand", "status",
    "base_price", "market_price", "cost_price", "member_price", "product_tier_prices",
    "stock_qty", "low_stock_threshold", "allow_oversell",
    "weight_grams", "length", "width", "height",
    "shelf_life", "shelf_life_en", "expiry_date",
    "description", "description_en", "ai_description", "ai_description_en",
    "meta_title", "meta_description", "seo_keywords",
    "meta_title_en", "meta_description_en", "seo_keywords_en",
    "cover_url", "images_json", "extra_attributes", "schema_markup",
]


def group_rows(rows: list[dict]) -> tuple[list[dict], list[str]]:
    """按 product_id（无则 product_sku）分组；返回 (groups, errors)。

    每组：{"key","product_id","product":公共字段, "variants":[行], "errors":[...]}。
    同组公共字段冲突 → 记入 errors。
    """
    groups: dict[str, dict] = {}
    errors: list[str] = []
    for i, r in enumerate(rows):
        pid = (r.get("product_id") or "").strip()
        psku = (r.get("product_sku") or "").strip()
        if not pid and not psku:
            errors.append(f"第{i + 1}行：缺少 product_id 与 product_sku，无法分组")
            continue
        key = f"id:{pid}" if pid else f"sku:{psku}"
        g = groups.get(key)
        common = {c: r.get(c, "") for c in _PRODUCT_COMMON}
        if g is None:
            groups[key] = {
                "key": key, "product_id": pid or None, "product_sku": psku,
                "product": common, "variants": [], "errors": [],
            }
            g = groups[key]
        else:
            for c in _PRODUCT_COMMON:
                if g["product"].get(c, "") != common.get(c, ""):
                    g["errors"].append(f"{key}: 公共字段 {c} 在多行间冲突")
                    break
        if (r.get("variant_sku") or "").strip() or (r.get("variant_id") or "").strip():
            g["variants"].append(r)
    return list(groups.values()), errors


# ── 值解析 ──────────────────────────────────────────────────────────
def _dec(v, field, errs):
    v = (v or "").strip()
    if v == "":
        return None
    try:
        return Decimal(v)
    except (InvalidOperation, ValueError):
        errs.append(f"{field} 非法数值 {v!r}")
        return None


def _int(v):
    v = (v or "").strip()
    return int(v) if v.lstrip("-").isdigit() else None


def _int_or(v, default: int) -> int:
    """空值取 default，但合法的 0 保留（避免 `_int(...) or default` 把 0 吞掉）。"""
    n = _int(v)
    return default if n is None else n


def _bool(v):
    return (v or "").strip() in ("1", "true", "True")


def _check_int(v, field: str, errs: list[str], *, minimum: int | None = None) -> None:
    text = (v or "").strip()
    if not text:
        return
    value = _int(text)
    if value is None:
        errs.append(f"{field} 非法整数 {text!r}")
    elif minimum is not None and value < minimum:
        errs.append(f"{field} 不能小于 {minimum}")


def _check_bool(v, field: str, errs: list[str]) -> None:
    text = (v or "").strip()
    if text and text not in ("0", "1", "false", "true", "False", "True"):
        errs.append(f"{field} 非法布尔值 {text!r}")


def _jsonload(v, field, errs):
    v = (v or "").strip()
    if v == "":
        return None
    try:
        return json.loads(v)
    except json.JSONDecodeError:
        errs.append(f"{field} 非法 JSON")
        return None


def _category_paths(raw) -> list[str]:
    """category_path（JSON 列表；兼容旧单路径字符串）→ 完整路径列表。"""
    raw = (raw or "").strip()
    if not raw:
        return []
    try:
        val = json.loads(raw)
        paths = val if isinstance(val, list) else [str(val)]
    except json.JSONDecodeError:
        paths = [raw]
    return [str(x).strip() for x in paths if str(x).strip()]


def _path_segments(path: str) -> list[str]:
    return [s.strip() for s in path.split("/") if s.strip()]


async def _load_category_index(db: AsyncSession, tenant_id: int) -> dict:
    """(parent_id, name) → id，用于按完整路径解析分类。"""
    rows = (await db.execute(
        select(Category.id, Category.parent_id, Category.name).where(Category.tenant_id == tenant_id)
    )).all()
    return {(parent_id, name): cid for cid, parent_id, name in rows}


def _resolve_path(index: dict, path: str) -> int | None:
    """按完整路径逐级解析到唯一分类 id；任一级缺失返回 None。"""
    parent = None
    cur = None
    for name in _path_segments(path):
        cur = index.get((parent, name))
        if cur is None:
            return None
        parent = cur
    return cur


async def _resolve_category_ids(db: AsyncSession, tenant_id: int, raw, create_missing: bool) -> list[int]:
    """完整路径列表 → 分类 id 列表；create_missing 时按完整父链逐级创建。

    按 (parent_id, name) 精确匹配，避免同名叶子（护肤/面霜 与 礼品/面霜）折叠或错配。
    """
    ids: list[int] = []
    for path in _category_paths(raw):
        parent = None
        cur = None
        ok = True
        for name in _path_segments(path):
            row = (await db.execute(
                select(Category.id).where(
                    Category.tenant_id == tenant_id, Category.name == name,
                    Category.parent_id == parent,
                ).limit(1)
            )).scalar_one_or_none()
            if row is None:
                if not create_missing:
                    ok = False
                    break
                # ponytail: slug 以 name-parent 保证 (tenant,slug) 唯一，够用；需要 SEO 友好 slug 再改
                c = Category(tenant_id=tenant_id, name=name, slug=f"{name}-{parent or 0}"[:200], parent_id=parent)
                db.add(c)
                await db.flush()
                row = c.id
            parent = row
            cur = row
        if ok and cur is not None and cur not in ids:
            ids.append(cur)
    return ids


# ── 只读预检 ────────────────────────────────────────────────────────
async def preflight(db: AsyncSession, tenant_id: int, rows: list[dict]) -> dict:
    """任何写入前完成的只读校验汇总。"""
    groups, errors = group_rows(rows)

    # 现有商品/规格/图片归属映射（均限当前租户），用于跨租户与归属校验
    prod_sku_map = dict((sku, pid) for sku, pid in (await db.execute(
        select(Product.sku, Product.id).where(Product.tenant_id == tenant_id)
    )).all() if sku)
    prod_ids = set((await db.execute(
        select(Product.id).where(Product.tenant_id == tenant_id)
    )).scalars().all())
    var_owner = {vid: (pid, sku) for vid, pid, sku in (await db.execute(
        select(ProductVariant.id, ProductVariant.product_id, ProductVariant.sku)
        .where(ProductVariant.tenant_id == tenant_id)
    )).all()}
    img_owner = dict((iid, pid) for iid, pid in (await db.execute(
        select(ProductImage.id, ProductImage.product_id).where(ProductImage.tenant_id == tenant_id)
    )).all())

    var_sku_map = {sku: (pid, vid) for vid, (pid, sku) in var_owner.items() if sku}

    missing_categories: set[str] = set()
    missing_brands: set[str] = set()
    cat_index = await _load_category_index(db, tenant_id)  # 按完整路径解析
    brand_names = set((await db.execute(
        select(Brand.name).where(Brand.tenant_id == tenant_id)  # 限本租户，避免跨租户品牌污染
    )).scalars().all())

    image_url_counts: dict[tuple[int, str], int] = {}
    for pid, url in (await db.execute(
        select(ProductImage.product_id, ProductImage.url).where(ProductImage.tenant_id == tenant_id)
    )).all():
        if url:
            key = (pid, url)
            image_url_counts[key] = image_url_counts.get(key, 0) + 1

    file_variant_skus: dict[str, str] = {}
    for g in groups:
        errors.extend(g["errors"])
        raw_pid = (g["product_id"] or "").strip()
        gpid = _int(raw_pid) if raw_pid else None
        if raw_pid and gpid is None:
            errors.append(f"{g['key']}: 商品 ID {raw_pid!r} 非法（应为数字）")
        if gpid is not None and gpid not in prod_ids:
            errors.append(f"{g['key']}: 商品 ID {gpid} 不存在或不属于当前租户")
        # SKU 与 ID 归属一致性（用 prod_sku_map 检测冲突）
        sku = (g["product"].get("product_sku") or "").strip()
        if sku and gpid is not None:
            owner = prod_sku_map.get(sku)
            if owner is not None and owner != gpid:
                errors.append(f"{g['key']}: SKU {sku} 已属于商品 {owner}，与商品 ID {gpid} 冲突")
        # 有效目标商品 id：优先文件 ID，否则按 SKU 定位现有商品（无则新建 → None）
        eff_pid = gpid if gpid is not None else prod_sku_map.get(sku)
        # 分类（按完整路径解析，避免同名叶子折叠/错配）与品牌
        for path in _category_paths(g["product"].get("category_path")):
            if _resolve_path(cat_index, path) is None:
                missing_categories.add(path)
        brand = (g["product"].get("brand") or "").strip()
        if brand and brand not in brand_names:
            missing_brands.add(brand)
        # 数值/JSON 合法性
        cell_errs: list[str] = []
        status = (g["product"].get("status") or "draft").strip()
        if status not in ("draft", "active", "archived"):
            cell_errs.append(f"{g['key']}.status 非法状态 {status!r}")
        for f in ("base_price", "market_price", "cost_price", "member_price", "stock_qty",
                  "length", "width", "height"):
            _dec(g["product"].get(f), f"{g['key']}.{f}", cell_errs)
        for f in ("low_stock_threshold", "weight_grams"):
            _check_int(g["product"].get(f), f"{g['key']}.{f}", cell_errs, minimum=0)
        _check_bool(g["product"].get("allow_oversell"), f"{g['key']}.allow_oversell", cell_errs)
        for f in ("category_path", "product_tier_prices", "images_json", "extra_attributes", "schema_markup"):
            _jsonload(g["product"].get(f), f"{g['key']}.{f}", cell_errs)
        errors.extend(cell_errs)
        # 图片校验：ID 归属、重复 URL、多主图、非法排序
        imgs = _jsonload(g["product"].get("images_json"), "", [])
        if imgs is not None and not isinstance(imgs, list):
            errors.append(f"{g['key']}: images_json 必须是列表")
        elif isinstance(imgs, list):
            seen_urls: set[str] = set()
            primary_count = 0
            for im in imgs:
                if not isinstance(im, dict):
                    errors.append(f"{g['key']}: 图片记录格式非法")
                    continue
                iid = im.get("id")
                if iid is not None:
                    if not isinstance(iid, int) or isinstance(iid, bool):
                        errors.append(f"{g['key']}: 图片 ID {iid!r} 非法（应为整数）")
                    elif iid not in img_owner:
                        errors.append(f"{g['key']}: 图片 ID {iid} 不存在或不属于当前租户")
                    elif eff_pid is not None and img_owner[iid] != eff_pid:
                        errors.append(f"{g['key']}: 图片 ID {iid} 不属于该商品")
                url = (im.get("url") or "").strip()
                if not url:
                    errors.append(f"{g['key']}: 图片缺少 URL")
                elif url in seen_urls:
                    errors.append(f"{g['key']}: 图片 URL 重复 {url}")
                else:
                    seen_urls.add(url)
                    if iid is None and eff_pid is not None and image_url_counts.get((eff_pid, url), 0) > 1:
                        errors.append(f"{g['key']}: 图片 URL {url} 无法唯一匹配")
                so = im.get("sort_order", 0)
                if not isinstance(so, int) or so < 0:
                    errors.append(f"{g['key']}: 图片排序非法 {so!r}")
                primary = im.get("is_primary", 0)
                if primary not in (0, 1, False, True, "0", "1", ""):
                    errors.append(f"{g['key']}: 图片主图标记非法 {primary!r}")
                elif int(primary or 0) == 1:
                    primary_count += 1
            if primary_count > 1:
                errors.append(f"{g['key']}: 存在多个主图")
        # 变体校验
        seen_variant_ids: set[int] = set()
        for v in g["variants"]:
            raw_vid = (v.get("variant_id") or "").strip()
            vid = _int(raw_vid) if raw_vid else None
            if raw_vid and vid is None:
                errors.append(f"{g['key']}: 规格 ID {raw_vid!r} 非法（应为数字）")
            if vid is not None:
                owner = var_owner.get(vid)
                if owner is None:
                    errors.append(f"{g['key']}: 规格 ID {vid} 不存在或不属于当前租户")
                elif eff_pid is not None and owner[0] != eff_pid:
                    errors.append(f"{g['key']}: 规格 ID {vid} 不属于该商品")
                if vid in seen_variant_ids:
                    errors.append(f"{g['key']}: 规格 ID {vid} 在文件中重复")
                seen_variant_ids.add(vid)
            # 数据库及本文件内的规格 SKU 均必须唯一
            vsku = (v.get("variant_sku") or "").strip()
            if vsku:
                sku_owner = var_sku_map.get(vsku)
                if sku_owner is not None:
                    owner_pid, owner_vid = sku_owner
                    if owner_pid != eff_pid or (vid is not None and owner_vid != vid):
                        errors.append(f"{g['key']}: 规格 SKU {vsku} 已属于规格 {owner_vid}")
                previous_group = file_variant_skus.get(vsku)
                if previous_group is not None:
                    errors.append(f"{g['key']}: 规格 SKU {vsku} 在文件中重复（首次见于 {previous_group}）")
                else:
                    file_variant_skus[vsku] = g["key"]
            vcell: list[str] = []
            for f in ("price_modifier", "variant_member_price", "variant_stock_qty", "reserved_qty",
                      "variant_length", "variant_width", "variant_height"):
                _dec(v.get(f), f"{g['key']}.{v.get('variant_sku')}.{f}", vcell)
            for f in ("variant_weight_grams", "sort_order"):
                _check_int(v.get(f), f"{g['key']}.{vsku}.{f}", vcell, minimum=0)
            for f in ("is_active", "is_default"):
                _check_bool(v.get(f), f"{g['key']}.{vsku}.{f}", vcell)
            for f in ("variant_tier_prices", "attributes"):
                _jsonload(v.get(f), f"{g['key']}.{v.get('variant_sku')}.{f}", vcell)
            errors.extend(vcell)

        # 复用核心输入模型验证日期、长度、非负库存和 JSON 结构；只构造模型，不写数据库。
        # ArithmeticError 覆盖 Decimal 非法值（如 tier price="abc" 抛的 InvalidOperation），
        # 否则会外抛导致预检以晦涩错误崩溃而非干净汇总。
        try:
            ProductUpdate(**_product_fields(g), variants=[_variant_in_from_row(v) for v in g["variants"]])
        except (ValidationError, TypeError, ValueError, ArithmeticError, AttributeError) as exc:
            errors.append(f"{g['key']}: 商品或规格字段非法：{str(exc)[:300]}")

    return {
        "groups": len(groups),
        "product_count": len(groups),
        "variant_rows": sum(len(g["variants"]) for g in groups),
        "missing_categories": sorted(missing_categories),
        "missing_brands": sorted(missing_brands),
        "errors": errors,
        "can_commit": len(errors) == 0,
    }


# ── 提交：payload 构造 ──────────────────────────────────────────────
def _tier_in(raw_json) -> list[TierPriceIn]:
    data = json.loads(raw_json) if raw_json else []
    return [TierPriceIn(member_level_id=int(t["member_level_id"]), price=Decimal(str(t["price"]))) for t in data]


def _variant_in_from_row(v: dict) -> ProductVariantIn:
    """文件行 → ProductVariantIn（不带 id，交由匹配阶段决定新增/更新）。"""
    grams = _int(v.get("variant_weight_grams"))
    return ProductVariantIn(
        sku=(v.get("variant_sku") or "").strip(),
        barcode=(v.get("barcode") or "").strip() or None,
        price_modifier=_dec(v.get("price_modifier"), "", []) or Decimal("0"),
        stock_qty=_dec(v.get("variant_stock_qty"), "", []) or Decimal("0"),
        reserved_qty=_dec(v.get("reserved_qty"), "", []) or Decimal("0"),
        image_url=(v.get("variant_image_url") or "").strip() or None,
        weight_grams=grams,
        length=_dec(v.get("variant_length"), "", []),
        width=_dec(v.get("variant_width"), "", []),
        height=_dec(v.get("variant_height"), "", []),
        member_price=_dec(v.get("variant_member_price"), "", []),
        sort_order=_int(v.get("sort_order")) or 0,
        attributes=json.loads(v["attributes"]) if (v.get("attributes") or "").strip() else {},
        is_active=_bool(v.get("is_active")),
        is_default=_bool(v.get("is_default")),
        expiry_date=(v.get("variant_expiry_date") or "").strip() or None,
        tier_prices=_tier_in(v.get("variant_tier_prices")),
    )


def _existing_variant_in(v: ProductVariant, vtiers: list[ProductTierPrice]) -> ProductVariantIn:
    """现有规格 → 完整 ProductVariantIn（含 id 与现有 tier_prices），用于合并保留。"""
    return ProductVariantIn(
        id=v.id, sku=v.sku, barcode=v.barcode,
        price_modifier=v.price_modifier or Decimal("0"),
        stock_qty=v.stock_qty or Decimal("0"), reserved_qty=v.reserved_qty or Decimal("0"),
        image_url=v.image_url,
        weight_grams=int((v.weight or 0) * 1000) if v.weight is not None else None,
        member_price=v.member_price,
        length=v.length, width=v.width, height=v.height,
        sort_order=v.sort_order, attributes=v.attributes or {},
        is_active=bool(v.is_active), is_default=bool(v.is_default), expiry_date=v.expiry_date,
        tier_prices=[TierPriceIn(member_level_id=t.member_level_id, price=t.price) for t in vtiers],
    )


def _product_fields(g: dict) -> dict:
    """商品公共字段 → create/update 通用字段 dict。"""
    p = g["product"]
    errs: list[str] = []
    grams = _int(p.get("weight_grams"))
    return {
        "name": (p.get("name") or "").strip() or None,
        "sku": (p.get("product_sku") or "").strip() or None,
        "name_en": (p.get("name_en") or "").strip() or None,
        "slug": (p.get("slug") or "").strip() or None,
        # categories 由 apply_group 按完整路径解析为 id 后注入
        "status": (p.get("status") or "draft").strip() or "draft",
        "base_price": _dec(p.get("base_price"), "base_price", errs) or Decimal("0"),
        "market_price": _dec(p.get("market_price"), "", errs),
        "cost_price": _dec(p.get("cost_price"), "", errs),
        "member_price": _dec(p.get("member_price"), "", errs),
        "stock_qty": _dec(p.get("stock_qty"), "", errs) or Decimal("0"),
        "low_stock_threshold": _int_or(p.get("low_stock_threshold"), 5),  # 合法 0 保留
        "allow_oversell": _bool(p.get("allow_oversell")),
        "weight_grams": grams or 0,
        "length": _dec(p.get("length"), "", errs),
        "width": _dec(p.get("width"), "", errs),
        "height": _dec(p.get("height"), "", errs),
        "expiry_date": (p.get("expiry_date") or "").strip() or None,
        "shelf_life": (p.get("shelf_life") or "").strip() or None,
        "shelf_life_en": (p.get("shelf_life_en") or "").strip() or None,
        "description": p.get("description") or None,
        "description_en": p.get("description_en") or None,
        "ai_description": p.get("ai_description") or None,
        "ai_description_en": p.get("ai_description_en") or None,
        "meta_title": (p.get("meta_title") or "").strip() or None,
        "meta_description": (p.get("meta_description") or "").strip() or None,
        "seo_keywords": (p.get("seo_keywords") or "").strip() or None,
        "meta_title_en": (p.get("meta_title_en") or "").strip() or None,
        "meta_description_en": (p.get("meta_description_en") or "").strip() or None,
        "seo_keywords_en": (p.get("seo_keywords_en") or "").strip() or None,
        "cover_url": (p.get("cover_url") or "").strip() or None,
        "tier_prices": _tier_in(p.get("product_tier_prices")),
    }


# ── 提交：落库（复用规范写入路径）─────────────────────────────────────
async def _resolve_brand_id(db, tenant_id, name, create_missing) -> int | None:
    name = (name or "").strip()
    if not name:
        return None
    bid = (await db.execute(
        select(Brand.id).where(Brand.tenant_id == tenant_id, Brand.name == name).limit(1)
    )).scalar_one_or_none()
    if bid or not create_missing:
        return bid
    from app.plugins.catalog_transfer.exporter import _s  # noqa
    b = Brand(tenant_id=tenant_id, name=name, slug=f"{name}"[:180])
    db.add(b)
    await db.flush()
    return b.id


async def _merge_variants(db, tenant_id, product_id, file_variants: list[dict]) -> list[ProductVariantIn]:
    """现有规格 + 文件规格 → 完整 ProductVariantIn 列表，落实“缺行不删”。"""
    existing = (await db.execute(
        select(ProductVariant).where(
            ProductVariant.tenant_id == tenant_id, ProductVariant.product_id == product_id
        )
    )).scalars().all()
    tiers = (await db.execute(
        select(ProductTierPrice).where(
            ProductTierPrice.tenant_id == tenant_id, ProductTierPrice.product_id == product_id,
            ProductTierPrice.variant_id.isnot(None),
        )
    )).scalars().all()
    tier_by_variant: dict[int, list] = {}
    for t in tiers:
        tier_by_variant.setdefault(t.variant_id, []).append(t)

    by_id = {v.id: v for v in existing}
    by_sku = {v.sku: v for v in existing}
    merged: dict[int, ProductVariantIn] = {}   # 现有 id → 合并后
    news: list[ProductVariantIn] = []
    matched_ids: set[int] = set()

    for fv in file_variants:
        vid = _int(fv.get("variant_id")) if (fv.get("variant_id") or "").strip() else None
        target = by_id.get(vid) if vid is not None else by_sku.get((fv.get("variant_sku") or "").strip())
        vin = _variant_in_from_row(fv)
        if target is not None:
            vin.id = target.id
            merged[target.id] = vin
            matched_ids.add(target.id)
        else:
            news.append(vin)

    # 未在文件中出现的现有规格：完整保留
    for v in existing:
        if v.id not in matched_ids:
            merged[v.id] = _existing_variant_in(v, tier_by_variant.get(v.id, []))

    return list(merged.values()) + news


async def _reconcile_images(db, tenant_id, product_id, images_json: str, cover_url: str | None) -> None:
    """按 图片 ID → URL 幂等对账：只新增/更新、不因缺行删除；主图互斥并同步 cover_url。

    图片归属、重复 URL、多主图、非法排序均在预检阶段拦截，此处只做写入。
    """
    data = json.loads(images_json) if (images_json or "").strip() else []
    existing = (await db.execute(
        select(ProductImage).where(
            ProductImage.tenant_id == tenant_id, ProductImage.product_id == product_id
        )
    )).scalars().all()
    by_id = {im.id: im for im in existing}
    by_url = {im.url: im for im in existing}
    touched = list(existing)   # 主图互斥时遍历，避免 by_url 键随 url 变更而失配
    primary_url = None
    for item in data:
        url = (item.get("url") or "").strip()
        iid = item.get("id")
        im = by_id.get(iid) if iid is not None else by_url.get(url)  # ID 优先，其次 URL
        if im is None:
            if not url:
                continue
            im = ProductImage(
                product_id=product_id, tenant_id=tenant_id, url=url,
                alt_text=(item.get("alt_text") or None),
                sort_order=int(item.get("sort_order") or 0), is_primary=0,
            )
            db.add(im)
            by_url[url] = im
            touched.append(im)
        else:
            if url:
                im.url = url
            im.alt_text = item.get("alt_text") or None
            im.sort_order = int(item.get("sort_order") or 0)
        if int(item.get("is_primary") or 0) == 1:
            primary_url = im.url
    # 主图：仅文件显式指定时互斥切换；未指定则完整保留现状。
    if primary_url is not None:
        for im in touched:
            im.is_primary = 1 if im.url == primary_url else 0
        await db.execute(
            update(Product).where(Product.id == product_id, Product.tenant_id == tenant_id)
            .values(cover_url=primary_url)
        )
    await db.commit()


async def apply_group(db: AsyncSession, tenant_id: int, user, g: dict, create_missing: bool) -> dict:
    """落库一个商品分组。复用 create_product/update_product（各自内部 commit）。"""
    from app.api.routers.products import create_product, update_product

    fields = _product_fields(g)
    fields["brand_id"] = await _resolve_brand_id(db, tenant_id, g["product"].get("brand"), create_missing)
    # 按完整路径解析（或按需创建父链）为分类 id，避免同名叶子折叠/错配
    fields["categories"] = await _resolve_category_ids(db, tenant_id, g["product"].get("category_path"), create_missing)
    if db.in_transaction():
        await db.commit()  # 提交上面的 brand/category 预建

    gpid = _int(g["product_id"]) if g["product_id"] else None
    target_id = gpid
    if target_id is None and fields["sku"]:
        target_id = (await db.execute(
            select(Product.id).where(Product.tenant_id == tenant_id, Product.sku == fields["sku"]).limit(1)
        )).scalar_one_or_none()

    if target_id is not None:
        variants = await _merge_variants(db, tenant_id, target_id, g["variants"])
        body = ProductUpdate(**fields, variants=variants)
        await update_product(target_id, body, db, user)
        action = "updated"
    else:
        variants = [_variant_in_from_row(v) for v in g["variants"]]
        body = ProductCreate(**fields, variants=variants)
        out = await create_product(body, db, user)
        target_id = out.id
        action = "created"

    # extra_attributes / schema_markup 不在 ProductBase 内，create/update_product 不处理，直接写
    extra = _jsonload(g["product"].get("extra_attributes"), "", [])
    schema = _jsonload(g["product"].get("schema_markup"), "", [])
    if extra is not None or schema is not None:
        await db.execute(
            update(Product).where(Product.id == target_id, Product.tenant_id == tenant_id)
            .values(extra_attributes=extra, schema_markup=schema)
        )
        await db.commit()

    await _reconcile_images(db, tenant_id, target_id, g["product"].get("images_json"), fields.get("cover_url"))
    return {"action": action, "product_id": target_id, "sku": fields["sku"]}