"""导出：稳定查询 → 扁平化（一规格一行）→ XLSX/ZIP。

字段契约（FORMAT_VERSION / COLUMNS / JSON 编码 / 超长拆分）为导出与回导共用，
importer.py 从本模块导入，保证往返一致。
"""
from __future__ import annotations

import json
import math
import os
import zipfile
from datetime import date, datetime
from decimal import Decimal

from openpyxl import Workbook
from openpyxl.cell import WriteOnlyCell
from openpyxl.cell.cell import ILLEGAL_CHARACTERS_RE
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

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

FORMAT_VERSION = "catalog-transfer-v1"

# 单元格文本上限 32767，留余量按 32000 拆分为 <col>_part_2、_part_3…
CELL_CHUNK = 32000

# 文本格式列（避免科学计数法/精度损失）
TEXT_COLUMNS = {"product_id", "product_sku", "variant_id", "variant_sku", "barcode"}

# JSON 列（稳定编码，回导按 JSON 解析）
JSON_COLUMNS = {
    "category_path", "product_tier_prices", "variant_tier_prices", "attributes",
    "images_json", "extra_attributes", "schema_markup",
}

# 只读参考列：导入时忽略
READONLY_COLUMNS = {"sales_count", "created_at", "updated_at"}

# 列顺序即 XLSX 列顺序
COLUMNS: list[str] = [
    "format_version",
    # 标识
    "product_id", "product_sku", "variant_id", "variant_sku", "barcode",
    # 基础
    "name", "name_en", "slug", "category_path", "brand", "status",
    # 商品价格
    "base_price", "market_price", "cost_price", "member_price",
    # 规格价格
    "price_modifier", "variant_member_price", "product_tier_prices", "variant_tier_prices",
    # 库存
    "stock_qty", "variant_stock_qty", "reserved_qty", "low_stock_threshold", "allow_oversell",
    # 规格
    "attributes", "is_default", "is_active", "sort_order",
    # 物流
    "weight_grams", "length", "width", "height",
    "variant_weight_grams", "variant_length", "variant_width", "variant_height",
    # 效期
    "shelf_life", "shelf_life_en", "expiry_date", "variant_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", "variant_image_url", "images_json",
    # 扩展
    "extra_attributes", "schema_markup",
    # 只读参考
    "sales_count", "created_at", "updated_at",
]


# ── 值转换 ──────────────────────────────────────────────────────────
def _s(value) -> str:
    """标量 → 稳定字符串。None/空 → ""。"""
    if value is None:
        return ""
    if isinstance(value, bool):
        return "1" if value else "0"
    if isinstance(value, Decimal):
        return format(value.normalize(), "f")
    if isinstance(value, (date, datetime)):
        return value.isoformat()
    return str(value)


def _json(obj) -> str:
    """稳定 JSON：固定键顺序、UTF-8、紧凑分隔。空 → ""。"""
    if obj is None or obj == [] or obj == {}:
        return ""
    return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def _grams(kg) -> str:
    if kg is None:
        return ""
    return str(int(Decimal(str(kg)) * 1000))


def _tier_list(rows: list[ProductTierPrice]) -> list[dict]:
    """稳定排序的阶梯价列表。"""
    return [
        {"member_level_id": r.member_level_id, "price": _s(r.price)}
        for r in sorted(rows, key=lambda r: r.member_level_id)
    ]


# ── 分类完整路径 ────────────────────────────────────────────────────
class _CategoryPaths:
    """一次加载全部分类，构造 id → "根/…/叶" 路径。"""

    def __init__(self, rows):
        self._name = {r.id: r.name for r in rows}
        self._parent = {r.id: r.parent_id for r in rows}

    def path_of(self, cat_id: int | None) -> str:
        parts, seen = [], set()
        cur = cat_id
        while cur is not None and cur in self._name and cur not in seen:
            seen.add(cur)
            parts.append(self._name[cur])
            cur = self._parent.get(cur)
        return "/".join(reversed(parts))


async def _load_category_paths(db: AsyncSession, tenant_id: int) -> _CategoryPaths:
    rows = (await db.execute(
        select(Category.id, Category.parent_id, Category.name).where(Category.tenant_id == tenant_id)
    )).all()
    return _CategoryPaths(rows)


# ── 范围解析 ────────────────────────────────────────────────────────
async def _base_scope_select(db: AsyncSession, tenant_id: int, scope: str, product_ids, filt):
    """返回 select(Product.id) 应用范围与筛选，稳定按 id 升序。"""
    variant_expiry = (
        select(
            ProductVariant.product_id.label("product_id"),
            func.min(ProductVariant.expiry_date).label("variant_expiry_date"),
        )
        .where(ProductVariant.tenant_id == tenant_id)
        .group_by(ProductVariant.product_id)
        .subquery()
    )
    effective = effective_expiry_expr(Product.expiry_date, variant_expiry.c.variant_expiry_date)
    q = (
        select(Product.id)
        .select_from(Product)
        .outerjoin(variant_expiry, variant_expiry.c.product_id == Product.id)
        .where(Product.tenant_id == tenant_id)
    )
    if scope == "selected":
        # 服务端不信任前端 ID：再次按租户限定
        q = q.where(Product.id.in_(product_ids or [-1]))
    elif scope == "filtered":
        conds = await build_product_filter_conditions(
            db, tenant_id,
            keyword=filt.keyword, category=filt.category, category_id=filt.category_id, status=filt.status,
            expiry_status=filt.expiry_status, effective_expiry=effective,
            warning_days=await _warning_days(db, tenant_id),
        )
        if conds:
            q = q.where(*conds)
    # scope == "all"：仅租户限定
    return q.order_by(Product.id.asc())


async def _warning_days(db: AsyncSession, tenant_id: int) -> int:
    from app.core.models.tenant_settings import TenantSettings
    r = await db.execute(select(TenantSettings.extra).where(TenantSettings.tenant_id == tenant_id))
    extra = r.scalar_one_or_none() or {}
    try:
        return max(0, int(extra.get("expiry_warning_days", 30)))
    except (TypeError, ValueError):
        return 30


async def resolve_scope_ids(db: AsyncSession, tenant_id: int, req) -> list[int]:
    """范围内全部主商品 ID（稳定升序）。

    ponytail: 一次性取回全部 ID 列表（10 万约几 MB），随后分块加载完整数据；
    ID 量级超百万再换 keyset 分页。
    """
    q = await _base_scope_select(db, tenant_id, req.scope, req.product_ids, req.filter)
    return [r[0] for r in (await db.execute(q)).all()]


async def estimate(db: AsyncSession, tenant_id: int, req) -> dict:
    """主商品数、输出行数、文件数。仅用计数，不加载商品对象。"""
    ids = await resolve_scope_ids(db, tenant_id, req)
    product_count = len(ids)
    if not product_count:
        return {"product_count": 0, "row_count": 0, "file_count": 0}
    # 有规格的商品：规格数；无规格商品：占 1 行
    variant_count = 0
    products_with_variants = 0
    for chunk in _chunks(ids, 5000):
        vc = (await db.execute(
            select(func.count()).select_from(ProductVariant)
            .where(ProductVariant.tenant_id == tenant_id, ProductVariant.product_id.in_(chunk))
        )).scalar() or 0
        pwv = (await db.execute(
            select(func.count(func.distinct(ProductVariant.product_id)))
            .where(ProductVariant.tenant_id == tenant_id, ProductVariant.product_id.in_(chunk))
        )).scalar() or 0
        variant_count += vc
        products_with_variants += pwv
    row_count = variant_count + (product_count - products_with_variants)
    file_count = max(1, math.ceil(row_count / req.rows_per_file))
    return {"product_count": product_count, "row_count": row_count, "file_count": file_count}


def _chunks(seq, size):
    for i in range(0, len(seq), size):
        yield seq[i:i + size]


# ── 扁平化 ──────────────────────────────────────────────────────────
async def _flatten_chunk(db: AsyncSession, tenant_id: int, product_ids: list[int],
                         cat_paths: _CategoryPaths):
    """一批商品 → 有序行 dict 列表，规格按 (sort_order, id) 升序，无规格占一行。"""
    prods = (await db.execute(
        select(Product).where(Product.tenant_id == tenant_id, Product.id.in_(product_ids))
    )).scalars().all()
    prod_map = {p.id: p for p in prods}

    variants = (await db.execute(
        select(ProductVariant)
        .where(ProductVariant.tenant_id == tenant_id, ProductVariant.product_id.in_(product_ids))
        .order_by(ProductVariant.product_id, ProductVariant.sort_order, ProductVariant.id)
    )).scalars().all()
    var_by_product: dict[int, list[ProductVariant]] = {}
    for v in variants:
        var_by_product.setdefault(v.product_id, []).append(v)

    tiers = (await db.execute(
        select(ProductTierPrice).where(
            ProductTierPrice.tenant_id == tenant_id, ProductTierPrice.product_id.in_(product_ids)
        )
    )).scalars().all()
    product_tiers: dict[int, list] = {}
    variant_tiers: dict[int, list] = {}
    for t in tiers:
        if t.variant_id is None:
            product_tiers.setdefault(t.product_id, []).append(t)
        else:
            variant_tiers.setdefault(t.variant_id, []).append(t)

    images = (await db.execute(
        select(ProductImage)
        .where(ProductImage.tenant_id == tenant_id, ProductImage.product_id.in_(product_ids))
        .order_by(ProductImage.sort_order, ProductImage.id)
    )).scalars().all()
    img_by_product: dict[int, list] = {}
    for im in images:
        img_by_product.setdefault(im.product_id, []).append(im)

    brand_ids = {p.brand_id for p in prods if p.brand_id}
    brand_map = {}
    if brand_ids:
        brs = (await db.execute(select(Brand).where(Brand.id.in_(brand_ids)))).scalars().all()
        brand_map = {b.id: b.name for b in brs}

    cats = (await db.execute(
        select(product_categories.c.product_id, product_categories.c.category_id)
        .where(product_categories.c.product_id.in_(product_ids))
    )).all()
    prod_cats: dict[int, list[int]] = {}
    for pid, cid in cats:
        prod_cats.setdefault(pid, []).append(cid)
    for pid in prod_cats:
        prod_cats[pid].sort()  # 稳定排序，保证多分类往返一致

    rows = []
    for pid in product_ids:  # 保持 id 升序
        p = prod_map.get(pid)
        if p is None:
            continue
        base = _product_base(p, brand_map, cat_paths, prod_cats.get(pid, []),
                             product_tiers.get(pid, []), img_by_product.get(pid, []))
        vs = var_by_product.get(pid, [])
        if not vs:
            rows.append({**base, **_empty_variant()})
        else:
            for v in vs:
                rows.append({**base, **_variant_cols(v, variant_tiers.get(v.id, []))})
    return rows


def _product_base(p: Product, brand_map, cat_paths, cat_ids, ptiers, imgs) -> dict:
    return {
        "format_version": FORMAT_VERSION,
        "product_id": _s(p.id), "product_sku": _s(p.sku),
        "name": _s(p.name), "name_en": _s(p.name_en), "slug": _s(p.slug),
        # 全部分类的完整路径（JSON 列表，稳定排序），回导据此恢复所有分类
        "category_path": _json([cat_paths.path_of(c) for c in cat_ids]),
        "brand": _s(brand_map.get(p.brand_id)), "status": _s(p.status),
        "base_price": _s(p.base_price), "market_price": _s(p.market_price),
        "cost_price": _s(p.cost_price), "member_price": _s(p.member_price),
        "product_tier_prices": _json(_tier_list(ptiers)),
        "stock_qty": _s(p.stock_qty),
        "low_stock_threshold": _s(p.low_stock_threshold), "allow_oversell": _s(p.allow_oversell),
        "weight_grams": _grams(p.weight),
        "length": _s(p.length), "width": _s(p.width), "height": _s(p.height),
        "shelf_life": _s(p.shelf_life), "shelf_life_en": _s(p.shelf_life_en),
        "expiry_date": _s(p.expiry_date),
        "description": _s(p.description), "description_en": _s(p.description_en),
        "ai_description": _s(p.ai_description), "ai_description_en": _s(p.ai_description_en),
        "meta_title": _s(p.meta_title), "meta_description": _s(p.meta_description),
        "seo_keywords": _s(p.seo_keywords),
        "meta_title_en": _s(p.meta_title_en), "meta_description_en": _s(p.meta_description_en),
        "seo_keywords_en": _s(p.seo_keywords_en),
        "cover_url": _s(p.cover_url),
        "images_json": _json([
            {"id": im.id, "url": im.url, "alt_text": im.alt_text or "",
             "sort_order": im.sort_order, "is_primary": int(im.is_primary)}
            for im in imgs
        ]),
        "extra_attributes": _json(p.extra_attributes),
        "schema_markup": _json(p.schema_markup),
        "sales_count": _s(p.sales_count),
        "created_at": _s(p.created_at), "updated_at": _s(getattr(p, "updated_at", None)),
    }


def _variant_cols(v: ProductVariant, vtiers) -> dict:
    return {
        "variant_id": _s(v.id), "variant_sku": _s(v.sku), "barcode": _s(v.barcode),
        "price_modifier": _s(v.price_modifier), "variant_member_price": _s(v.member_price),
        "variant_tier_prices": _json(_tier_list(vtiers)),
        "variant_stock_qty": _s(v.stock_qty), "reserved_qty": _s(v.reserved_qty),
        "attributes": _json(v.attributes),
        "is_default": _s(v.is_default), "is_active": _s(v.is_active), "sort_order": _s(v.sort_order),
        "variant_weight_grams": _grams(v.weight),
        "variant_length": _s(v.length), "variant_width": _s(v.width), "variant_height": _s(v.height),
        "variant_expiry_date": _s(v.expiry_date),
        "variant_image_url": _s(v.image_url),
    }


def _empty_variant() -> dict:
    return {k: "" for k in (
        "variant_id", "variant_sku", "barcode", "price_modifier", "variant_member_price",
        "variant_tier_prices", "variant_stock_qty", "reserved_qty", "attributes",
        "is_default", "is_active", "sort_order", "variant_weight_grams",
        "variant_length", "variant_width", "variant_height", "variant_expiry_date",
        "variant_image_url",
    )}


# ── XLSX 写入 ───────────────────────────────────────────────────────
def _clean_cell(v):
    """剥除 XLSX 非法控制字符（\\x00-\\x08,\\x0b,\\x0c,\\x0e-\\x1f）；保留 \\t\\n\\r。"""
    return ILLEGAL_CHARACTERS_RE.sub("", v) if isinstance(v, str) else v


def _split_value(value: str) -> list[str]:
    if len(value) <= CELL_CHUNK:
        return [value]
    return [value[i:i + CELL_CHUNK] for i in range(0, len(value), CELL_CHUNK)]


def _build_headers(rows: list[dict]) -> tuple[list[str], dict[str, int]]:
    """按本文件各列最大分片数，生成含 _part_N 的表头。"""
    parts: dict[str, int] = {}
    for col in COLUMNS:
        mx = 1
        for r in rows:
            n = math.ceil(len(r.get(col, "")) / CELL_CHUNK) or 1
            if n > mx:
                mx = n
        parts[col] = mx
    headers = []
    for col in COLUMNS:
        headers.append(col)
        for i in range(2, parts[col] + 1):
            headers.append(f"{col}_part_{i}")
    return headers, parts


def write_xlsx(rows: list[dict], path: str) -> None:
    """将一个文件的行写入 XLSX（只写模式，控内存）：清洗非法控制字符 + 文本格式 +
    quotePrefix 防注入 + 超长拆分。"""
    # 剥除 XLSX 不允许的控制字符（如脏数据里的 \x02），否则 openpyxl 直接抛
    # IllegalCharacterError 使整批导出失败。控制字符本属脏数据，剥掉即可。
    rows = [{k: _clean_cell(v) for k, v in r.items()} for r in rows]
    headers, parts = _build_headers(rows)
    wb = Workbook(write_only=True)
    ws = wb.create_sheet("products")
    ws.append(headers)
    for r in rows:
        cells = []
        for col in COLUMNS:
            chunks = _split_value(r.get(col, ""))
            is_text = col in TEXT_COLUMNS
            for j in range(parts[col]):
                val = chunks[j] if j < len(chunks) else ""
                cell = WriteOnlyCell(ws, value=val)
                if is_text:
                    cell.number_format = "@"
                # 公式注入防护：= + - @ 开头强制字符串类型（阻止当公式）并加文本前缀，
                # 均不改变回读的真实值。
                if isinstance(val, str) and val[:1] in ("=", "+", "-", "@"):
                    cell.data_type = "s"
                    cell.quotePrefix = True
                cells.append(cell)
        ws.append(cells)
    wb.save(path)


def file_name(index: int, start_row: int, end_row: int) -> str:
    return f"products_{index:03d}_rows_{start_row}-{end_row}.xlsx"


def zip_name() -> str:
    return f"catalog_transfer_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.zip"


def pack_zip(file_paths: list[str], zip_path: str) -> None:
    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
        for p in file_paths:
            zf.write(p, arcname=os.path.basename(p))