"""表格批量入库：解析 xlsx/csv → 行 → 列映射 → 交给 ai.enrich_batch → commit。"""
import csv, io, re

# 从任意常见写法里抓出 年/月/日：2029/3/31、2029-03-31 00:00:00、2029--3-31 等
_DATE_RE = re.compile(r"(\d{4})\D+(\d{1,2})\D+(\d{1,2})")

def _normalize_date(val) -> str | None:
    m = _DATE_RE.search(str(val or ""))
    if not m:
        return None
    y, mo, d = m.groups()
    return f"{y}-{int(mo):02d}-{int(d):02d}"

# 目标字段 → 可能的表头别名（小写包含匹配）
# ponytail: 英文(_en)字段放最前，配合 guess 的 used 集合优先认领「英文xx」列，
#           避免被中文字段的子串（如「英文名称」含「名称」）抢走。
_ALIASES = {
    "name_en":        ["英文名", "英文名称", "name(en)", "name en", "name_en", "english name"],
    "description_en": ["英文描述", "英文详情", "description(en)", "description_en"],
    "shelf_life_en":  ["英文保质期", "shelf life(en)", "shelf_life_en"],
    "category_level_1": ["一级分类", "category level 1", "category_level_1"],
    "category_level_2": ["二级分类", "category level 2", "category_level_2"],
    "category_level_3": ["三级分类", "category level 3", "category_level_3"],
    "name":         ["品名", "名称", "商品名", "name", "title", "product"],
    "sku":          ["sku", "编码", "货号", "条码", "barcode"],
    "base_price":   ["售价", "价格", "单价", "price", "base_price"],
    "market_price": ["市场价", "划线价", "原价", "market"],
    "cost_price":   ["成本", "进价", "cost"],
    "member_price": ["会员价", "member"],
    "stock_qty":    ["库存", "数量", "stock", "qty", "quantity"],
    "weight_grams": ["重量", "克重", "weight"],
    "category":     ["分类路径", "category path", "分类", "类目", "category"],
    "brand":        ["品牌", "brand"],
    "description":  ["描述", "详情", "description"],
    "shelf_life":   ["保质期", "shelf life", "shelf"],
    "expiry_date":  ["到期", "有效期至", "过期", "expiry", "expire", "best before"],
}

def guess_column_map(headers: list[str]) -> dict:
    out = {}
    used = set()  # 一个表头只归一个字段，防止「英文名称」既进 name_en 又进 name
    for field, aliases in _ALIASES.items():
        for h in headers:
            if h in used:
                continue
            hl = str(h).strip().lower()
            if any(a in hl for a in aliases):
                out[field] = h
                used.add(h)
                break
    return out

def parse_csv(content: bytes) -> tuple[list[str], list[dict]]:
    text = content.decode("utf-8-sig")
    reader = csv.reader(io.StringIO(text))
    rows = list(reader)
    if not rows:
        return [], []
    headers = rows[0]
    data = [dict(zip(headers, r)) for r in rows[1:] if any(c.strip() for c in r)]
    return headers, data

_TEMPLATE_HEADERS = [
    "商品名称(必填)", "英文名称", "SKU(必填)", "售价", "市场价", "成本价", "会员价",
    "库存", "重量(克)", "分类路径", "一级分类", "二级分类", "三级分类", "品牌", "保质期", "英文保质期", "到期日期", "商品描述", "英文描述",
]

def build_template_xlsx() -> bytes:
    """生成 xlsx 模板：SKU 列预设为文本格式，避免 Excel 把长条码变成科学计数法丢精度。"""
    from openpyxl import Workbook  # 懒加载
    wb = Workbook()
    ws = wb.active
    ws.append(_TEMPLATE_HEADERS)
    ws.append(["示例：食品 > 饮料 > 茶；礼品 > 茶礼盒" if h == "分类路径" else "食品" if h == "一级分类" else "饮料" if h == "二级分类" else "茶" if h == "三级分类" else "" for h in _TEMPLATE_HEADERS])
    sku_col = _TEMPLATE_HEADERS.index("SKU(必填)") + 1  # 1-based
    letter = ws.cell(row=1, column=sku_col).column_letter
    # 预格式化表头以下若干行为文本，用户填条码时保持原样
    for r in range(1, 2001):
        ws[f"{letter}{r}"].number_format = "@"
    buf = io.BytesIO()
    wb.save(buf)
    return buf.getvalue()


def parse_xlsx(content: bytes) -> tuple[list[str], list[dict]]:
    from openpyxl import load_workbook  # 懒加载
    wb = load_workbook(io.BytesIO(content), read_only=True, data_only=True)
    ws = wb.active
    it = ws.iter_rows(values_only=True)
    headers = [str(c) if c is not None else "" for c in next(it)]
    data = []
    for r in it:
        if not any(c is not None and str(c).strip() for c in r):
            continue
        data.append({h: ("" if v is None else str(v)) for h, v in zip(headers, r)})
    return headers, data


def category_paths_from_row(row: dict, column_map: dict) -> tuple[list[list[str]], list[str]]:
    warnings = []
    level_fields = sorted(
        (
            (int(field.removeprefix("category_level_")), source)
            for field, source in column_map.items()
            if field.startswith("category_level_") and field.removeprefix("category_level_").isdigit()
        ),
        key=lambda pair: pair[0],
    )
    populated = [(level, str(row.get(source) or "").strip()) for level, source in level_fields]
    populated = [(level, value) for level, value in populated if value]
    path_source = column_map.get("category")
    path_value = str(row.get(path_source) or "").strip() if path_source else ""

    if populated:
        highest = populated[-1][0]
        values = {level: value for level, value in populated}
        if any(not values.get(level) for level in range(1, highest + 1)):
            return [], ["category levels cannot continue after a blank level"]
        if path_value:
            warnings.append("category path column ignored because level columns are populated")
        return [[values[level] for level in range(1, highest + 1)]], warnings

    paths = []
    for raw_path in re.split(r"[；;]", path_value):
        raw_path = raw_path.strip()
        if not raw_path:
            continue
        levels = [level.strip() for level in raw_path.split(">")]
        if any(not level for level in levels):
            return [], ["category path contains a blank level"]
        if levels not in paths:
            paths.append(levels)
    return paths, warnings

def rows_to_product_creates(rows: list[dict], col_map: dict) -> list[dict]:
    """把源表行按 col_map（目标字段->源列名）转成 ProductCreate 用的 dict。
    category 源列 -> categories=[值]（与手动建商品表单一致，按名称）。缺 base_price 的行标记 _needs_price。"""
    out = []
    for r in rows:
        d = {}
        category_paths, category_warnings = category_paths_from_row(r, col_map)
        if category_paths:
            d["category_paths"] = category_paths
            d["categories"] = [path[-1] for path in category_paths]
        if category_warnings:
            d["_category_warnings"] = category_warnings
        for field, src_col in col_map.items():
            if field == "category" or field.startswith("category_level_"):
                continue
            if field == "brand":
                val = str(r.get(src_col, "") or "").strip()
                if val:
                    d["_brand_name"] = val
            elif field == "expiry_date":
                nd = _normalize_date(r.get(src_col))  # 解析失败则丢弃该字段，不拖垮整行
                if nd:
                    d[field] = nd
            else:
                val = r.get(src_col)
                if val not in (None, ""):
                    d[field] = str(val).strip() if isinstance(val, str) else val
        if not str(d.get("base_price") or "").strip():
            d["_needs_price"] = True
        out.append(d)
    return out


def taxonomy_report(rows: list[dict], categories, brand_names) -> dict:
    by_id = {category.id: category for category in categories}
    existing_paths = set()

    def path_for(category):
        names = [category.name]
        parent_id = category.parent_id
        seen = {category.id}
        while parent_id is not None and parent_id in by_id and parent_id not in seen:
            parent = by_id[parent_id]
            names.append(parent.name)
            seen.add(parent_id)
            parent_id = parent.parent_id
        return tuple(reversed(names))

    for category in categories:
        existing_paths.add(path_for(category))

    missing_paths = []
    seen_missing = set()
    missing_brands = []
    seen_brands = set()
    existing_brands = set(brand_names)
    invalid_paths = []
    warnings = []

    for row in rows:
        for warning in row.get("_category_warnings") or []:
            target = invalid_paths if "blank level" in warning else warnings
            if warning not in target:
                target.append(warning)
        for path in row.get("category_paths") or []:
            prefix = []
            for name in path:
                prefix.append(name)
                key = tuple(prefix)
                if key not in existing_paths and key not in seen_missing:
                    missing_paths.append(list(key))
                    seen_missing.add(key)
        brand = str(row.get("_brand_name") or "").strip()
        if brand and brand not in existing_brands and brand not in seen_brands:
            missing_brands.append(brand)
            seen_brands.add(brand)

    return {
        "missing_category_paths": missing_paths,
        "missing_brands": missing_brands,
        "invalid_paths": invalid_paths,
        "warnings": warnings,
        "requires_confirmation": bool(missing_paths or missing_brands),
    }


async def preflight_taxonomy(db, tenant_id: int, rows: list[dict]) -> dict:
    from sqlalchemy import select
    from app.core.models.category import Category
    from app.core.models.brand import Brand

    categories = (await db.execute(
        select(Category).where(Category.tenant_id == tenant_id)
    )).scalars().all()
    brands = (await db.execute(
        select(Brand.name).where(Brand.tenant_id == tenant_id)
    )).scalars().all()
    return taxonomy_report(rows, categories, brands)

async def _ensure_category_path(db, tenant_id: int, path: list[str]) -> int:
    from sqlalchemy import select
    from slugify import slugify
    from app.core.models.category import Category

    parent_id = None
    full_path = []
    for name in path:
        full_path.append(name)
        category = (await db.execute(
            select(Category).where(
                Category.tenant_id == tenant_id,
                Category.parent_id == parent_id,
                Category.name == name,
            ).limit(1)
        )).scalar_one_or_none()
        if category is None:
            slug = slugify("-".join(full_path), allow_unicode=True, separator="-") or f"category-{tenant_id}"
            category = Category(
                tenant_id=tenant_id,
                parent_id=parent_id,
                name=name,
                slug=slug[:200],
                is_active=1,
                is_nav_visible=1,
            )
            db.add(category)
            await db.flush()
        parent_id = category.id
    return parent_id


async def prepare_taxonomy(db, tenant_id: int, rows: list[dict], create_missing: bool) -> list[dict]:
    report = await preflight_taxonomy(db, tenant_id, rows)
    if report["invalid_paths"]:
        raise ValueError("; ".join(report["invalid_paths"]))
    if report["requires_confirmation"] and not create_missing:
        raise ValueError("taxonomy creation confirmation required")

    seen_paths = {}
    for row in rows:
        category_ids = []
        for path in row.get("category_paths") or []:
            key = tuple(path)
            if key not in seen_paths:
                seen_paths[key] = await _ensure_category_path(db, tenant_id, path)
            category_ids.append(seen_paths[key])
        if category_ids:
            row["categories"] = category_ids

    seen_brands = set()
    for row in rows:
        brand = str(row.get("_brand_name") or "").strip()
        if brand and brand not in seen_brands:
            await _get_or_create_brand(db, tenant_id, brand)
            seen_brands.add(brand)
    return rows

async def _build_brand_cache(db, tenant_id: int) -> dict[str, int]:
    from sqlalchemy import select
    from app.core.models.brand import Brand
    rows = await db.execute(select(Brand.name, Brand.id).where(Brand.tenant_id == tenant_id))
    return {name: bid for name, bid in rows.all()}


async def _get_or_create_brand(db, tenant_id: int, name: str) -> int:
    from sqlalchemy import select
    from app.core.models.brand import Brand
    row = await db.execute(select(Brand.id).where(Brand.tenant_id == tenant_id, Brand.name == name).limit(1))
    bid = row.scalar_one_or_none()
    if bid:
        return bid
    brand = Brand(tenant_id=tenant_id, name=name, slug=name.lower().replace(" ", "-"))
    db.add(brand)
    await db.commit()
    await db.refresh(brand)
    return brand.id


async def find_existing_skus(db, tenant_id: int, skus: list[str]) -> dict:
    """返回本租户下已存在的 {sku: product_id}，用于重复检测与覆盖。"""
    from sqlalchemy import select
    from app.core.models.product import Product
    skus = [s for s in skus if s]
    if not skus:
        return {}
    rows = await db.execute(
        select(Product.sku, Product.id).where(Product.tenant_id == tenant_id, Product.sku.in_(skus))
    )
    return {sku: pid for sku, pid in rows.all()}


async def commit_rows(session_factory, current_user, creates: list[dict], mode: str = "skip", progress_cb=None,
                      existing: dict | None = None, brand_cache: dict | None = None) -> dict:
    """逐行入库，单行失败不影响其它行。
    session_factory: async_sessionmaker——每行开一个独立 session，与 create_product 的
      「一请求一 session 一商品」设计一致，避免复用同一 session 累积状态导致 greenlet 错误。
    mode: skip=跳过已存在SKU；overwrite=更新已存在商品。"""
    from app.schemas.product import ProductCreate, ProductUpdate
    from app.api.routers.products import create_product, update_product

    if existing is None or brand_cache is None:
        async with session_factory() as db:
            if existing is None:
                existing = await find_existing_skus(db, current_user.tenant_id, [c.get("sku") for c in creates])
            if brand_cache is None:
                brand_cache = await _build_brand_cache(db, current_user.tenant_id)

    created = updated = skipped = 0
    errors = []
    total = len(creates)
    for i, raw in enumerate(creates):
        clean = {k: v for k, v in raw.items() if k not in ("_needs_price", "_brand_name", "category_paths", "_category_warnings")}
        brand_name = raw.get("_brand_name")
        if brand_name:
            if brand_name not in brand_cache:
                async with session_factory() as db:
                    brand_cache[brand_name] = await _get_or_create_brand(db, current_user.tenant_id, brand_name)
            clean["brand_id"] = brand_cache[brand_name]
        sku = clean.get("sku")
        try:
            if not clean.get("name") or not sku:
                errors.append({"index": i, "error": "缺少 name 或 sku"})
            elif sku in existing:
                if mode == "overwrite":
                    async with session_factory() as db:  # 每行独立 session
                        await update_product(existing[sku], ProductUpdate(**clean), db, current_user)
                    updated += 1
                else:
                    skipped += 1
            else:
                async with session_factory() as db:  # 每行独立 session
                    created_out = await create_product(ProductCreate(**clean), db, current_user)
                existing[sku] = created_out.id  # 存真实id：同批次重复SKU再出现时可正确走overwrite
                created += 1
        except Exception as e:
            errors.append({"index": i, "error": str(e)[:200]})
        if progress_cb is not None:
            await progress_cb(i + 1, total)
    return {"created": created, "updated": updated, "skipped": skipped, "errors": errors}