"""catalog_transfer Celery 任务：导出、回导预检、回导提交。

状态/进度写入 AsyncTaskLog，供 GET /api/admin/catalog-transfer/tasks/{id} 轮询。
只共用公共基础设施（AsyncTaskLog、Celery、Redis、商品写入路径），不触碰 bulk_import。
"""
from __future__ import annotations

import asyncio
import os
import shutil
import tempfile
from contextlib import asynccontextmanager
from datetime import datetime, timezone, timedelta

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

from app.celery_app import celery_app
from app.config import settings
from app.core.models.task import AsyncTaskLog

from . import exporter, importer
from .schemas import ExportRequest

RETENTION_HOURS = 24
BASE_DIR = os.path.join(tempfile.gettempdir(), "catalog_transfer")


# ── 任务会话与状态 ───────────────────────────────────────────────────
@asynccontextmanager
async def _task_engine():
    engine = create_async_engine(settings.DATABASE_URL, pool_pre_ping=True)
    try:
        yield async_sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False, autoflush=False)
    finally:
        await engine.dispose()


async def _set_task(task_id: int, session_factory, *, status: str | None = None, result: dict | None = None) -> None:
    async with session_factory() as db:
        row = await db.get(AsyncTaskLog, task_id)
        if row is None:
            return
        if status is not None:
            row.status = status
            if status in ("success", "failed", "cancelled"):
                row.completed_at = datetime.now(timezone.utc)
        if result is not None:
            row.result = result
        await db.commit()


async def _mark_failed(task_id: int, exc: Exception) -> None:
    async with _task_engine() as SF:
        await _set_task(task_id, SF, status="failed", result={"error": str(exc)[:300]})


# ── 文件目录与清理 ───────────────────────────────────────────────────
def task_dir(task_id: int) -> str:
    return os.path.join(BASE_DIR, str(task_id))


def cleanup_expired() -> None:
    """机会式清理：删除超过保留期的任务目录（按目录 mtime）。

    ponytail: 无定时服务，创建/查询/下载时顺带扫描；磁盘增长证明不足再加周期任务。
    """
    if not os.path.isdir(BASE_DIR):
        return
    cutoff = datetime.now().timestamp() - RETENTION_HOURS * 3600
    for name in os.listdir(BASE_DIR):
        p = os.path.join(BASE_DIR, name)
        try:
            if os.path.isdir(p) and os.path.getmtime(p) < cutoff:
                shutil.rmtree(p, ignore_errors=True)
        except OSError:
            pass


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


# ── 导出 ────────────────────────────────────────────────────────────
@celery_app.task(bind=True, name="catalog_transfer.export", acks_late=True)
def export_task(self, *, task_id: int, tenant_id: int, request: dict):
    async def run():
        async with _task_engine() as SF:
            req = ExportRequest(**request)
            work_dir = task_dir(task_id)
            os.makedirs(work_dir, exist_ok=True)
            buffer: list[dict] = []
            files: list[str] = []
            idx = 0
            row_start = 1
            produced = 0
            done = 0

            def flush():
                nonlocal idx, row_start, buffer
                if not buffer:
                    return
                idx += 1
                path = os.path.join(work_dir, exporter.file_name(idx, row_start, row_start + len(buffer) - 1))
                exporter.write_xlsx(buffer, path)
                files.append(path)
                row_start += len(buffer)
                buffer = []

            # 单一只读会话贯穿全部读取 → MySQL RR 一致性快照，避免漏行/重复
            # ponytail: 长事务会累积 undo log；10 万级可接受，更大再改分段快照+版本号核对
            async with SF() as db:
                ids = await exporter.resolve_scope_ids(db, tenant_id, req)
                cat_paths = await exporter._load_category_paths(db, tenant_id)
                total = len(ids)
                await _set_task(task_id, SF, status="running", result={"progress": {"done": 0, "total": total}})
                for chunk in _chunks(ids, 500):
                    rows = await exporter._flatten_chunk(db, tenant_id, chunk, cat_paths)
                    for r in rows:
                        buffer.append(r)
                        produced += 1
                        if len(buffer) >= req.rows_per_file:
                            flush()
                    done += len(chunk)
                    await _set_task(task_id, SF, result={"progress": {"done": done, "total": total}})
            flush()

            if len(files) <= 1:
                download = files[0] if files else None
            else:
                zpath = os.path.join(work_dir, exporter.zip_name())
                exporter.pack_zip(files, zpath)
                download = zpath

            expires = (datetime.now(timezone.utc) + timedelta(hours=RETENTION_HOURS)).isoformat()
            await _set_task(task_id, SF, status="success", result={
                "progress": {"done": total, "total": total},
                "row_count": produced, "file_count": len(files),
                "download_path": download,
                "download_name": os.path.basename(download) if download else None,
                "expires_at": expires,
            })

    try:
        asyncio.run(run())
    except Exception as exc:
        asyncio.run(_mark_failed(task_id, exc))
        raise


# ── 回导预检 ────────────────────────────────────────────────────────
@celery_app.task(bind=True, name="catalog_transfer.import_preflight", acks_late=True)
def preflight_task(self, *, task_id: int, tenant_id: int):
    async def run():
        async with _task_engine() as SF:
            async with SF() as db:
                parent = await db.get(AsyncTaskLog, task_id)
                payload = (parent.payload if parent else {}) or {}
            path, filename = payload.get("upload_path"), payload.get("filename", "")
            await _set_task(task_id, SF, status="running", result={"stage": "preflight"})
            with open(path, "rb") as f:
                data = f.read()
            try:
                rows = importer.read_input(data, filename)
                async with SF() as db:
                    report = await importer.preflight(db, tenant_id, rows)
            except importer.ImportError_ as e:
                report = {"errors": e.errors, "can_commit": False}
            await _set_task(task_id, SF, status="success", result={"preflight": report})

    try:
        asyncio.run(run())
    except Exception as exc:
        asyncio.run(_mark_failed(task_id, exc))
        raise


# ── 回导提交 ────────────────────────────────────────────────────────
@celery_app.task(bind=True, name="catalog_transfer.import_commit", acks_late=True)
def commit_task(self, *, task_id: int, tenant_id: int, user_id: int, create_missing: bool = False):
    async def run():
        from app.core.models.user import User

        async with _task_engine() as SF:
            async with SF() as db:
                parent = await db.get(AsyncTaskLog, task_id)
                payload = (parent.payload if parent else {}) or {}
                user = await db.get(User, user_id)
                if user is None or user.tenant_id != tenant_id:
                    raise ValueError("import user not found")
            with open(payload["upload_path"], "rb") as f:
                data = f.read()
            rows = importer.read_input(data, payload.get("filename", ""))
            async with SF() as db:
                report = await importer.preflight(db, tenant_id, rows)
                if not report.get("can_commit"):
                    await _set_task(task_id, SF, status="failed", result={"preflight": report, "error": "预检未通过"})
                    return
                # 防御：缺失分类/品牌且未确认创建时拒绝，避免静默清空
                if not create_missing and (report.get("missing_categories") or report.get("missing_brands")):
                    await _set_task(task_id, SF, status="failed",
                                    result={"preflight": report, "error": "存在缺失分类/品牌，未确认创建"})
                    return
            groups, _ = importer.group_rows(rows)

            totals = {"created": 0, "updated": 0, "failed": 0, "errors": []}
            await _set_task(task_id, SF, status="running", result={**totals, "progress": {"done": 0, "total": len(groups)}})
            done = 0
            for g in groups:
                async with SF() as db:
                    u = await db.get(User, user_id)
                    try:
                        res = await importer.apply_group(db, tenant_id, u, g, create_missing)
                        totals[res["action"]] += 1
                    except Exception as exc:  # 单组失败：记录，不回滚已提交，不假成功
                        totals["failed"] += 1
                        totals["errors"].append({"key": g["key"], "error": str(exc)[:200]})
                done += 1
                await _set_task(task_id, SF, result={**totals, "progress": {"done": done, "total": len(groups)}})

            status = "failed" if totals["failed"] else "success"
            await _set_task(task_id, SF, status=status, result={**totals, "progress": {"done": done, "total": len(groups)}})

    try:
        asyncio.run(run())
    except Exception as exc:
        asyncio.run(_mark_failed(task_id, exc))
        raise