"""
定时任务：从 ODPS 同步全局树元素与分类到 MySQL。

流程：
1. 拉取 pattern_mining_element（name, category_id）
2. 拉取 public_pattern_mining_category 全量分类
3. 筛选元素关联的分类，并沿 parent_id 向上追溯至顶层
4. 查询 MySQL 已有分类，按 source_id 去重（仅插入新增）
5. 为新分类分配 MySQL id/parent_id，INSERT IGNORE 写入（历史分类不变）
6. 重载 source_id 映射，将元素写入 global_tree_element（INSERT IGNORE 去重）
"""
from __future__ import annotations

import logging
from datetime import datetime, timedelta
from typing import Any

from supply_infra.db.repositories.global_tree_category_repo import GlobalTreeCategoryRepository
from supply_infra.db.repositories.global_tree_element_repo import GlobalTreeElementRepository
from supply_infra.db.session import get_session
from supply_infra.odps.client import get_odps_client

logger = logging.getLogger(__name__)


def _collect_categories_with_ancestors(
    seed_ids: set[int],
    category_by_id: dict[int, dict[str, Any]],
) -> list[dict[str, Any]]:
    """从种子 category_id 出发，沿 parent_id 向上追溯，收集完整祖先链。"""
    collected: dict[int, dict[str, Any]] = {}
    pending = set(seed_ids)

    while pending:
        cat_id = pending.pop()
        if cat_id in collected:
            continue

        category = category_by_id.get(cat_id)
        if category is None:
            continue

        collected[cat_id] = category
        parent_id = category.get("parent_id")
        if parent_id is not None and int(parent_id) != 0 and int(parent_id) not in collected:
            pending.add(int(parent_id))

    return list(collected.values())


def _dedupe_category_source_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """按 source_id 去重，保留最后一条（同日 ODPS 数据可能重复）。"""
    by_source_id: dict[int, dict[str, Any]] = {}
    for row in rows:
        by_source_id[int(row["source_id"])] = row
    return list(by_source_id.values())


def _dedupe_element_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """按 (name, category_id) 去重，与表唯一键一致。"""
    seen: set[tuple[str, int]] = set()
    result: list[dict[str, Any]] = []
    for row in rows:
        key = (row["name"], row["category_id"])
        if key in seen:
            continue
        seen.add(key)
        result.append(row)
    return result


def _to_category_source_rows(categories: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """将 ODPS 分类转为含 source_id / source_parent_id 的中间结构。"""
    rows: list[dict[str, Any]] = []
    for cat in categories:
        parent_id = cat.get("parent_id")
        rows.append(
            {
                "source_id": int(cat["id"]),
                "source_parent_id": int(parent_id) if parent_id is not None else 0,
                "name": str(cat["name"]) if cat.get("name") is not None else None,
                "description": cat.get("description"),
                "level": int(cat["level"]) if cat.get("level") is not None else None,
            }
        )
    return rows


def _prepare_new_category_rows(
    categories: list[dict[str, Any]],
    source_id_to_mysql_id: dict[int, int],
    next_id: int,
) -> tuple[list[dict[str, Any]], dict[int, int], int]:
    """
    过滤已存在 source_id，多轮扫描为新分类分配 MySQL id 与 parent_id。

    按 level 初排后多轮插入：仅当父节点已在映射中（库内已有或本轮已分配）才写入，
    避免同批次内因 level 数据异常导致子节点被误跳过。

    Returns:
        (待插入行, 更新后的 source_id → mysql_id 映射, 下一可用 id)
    """
    existing_source_ids = set(source_id_to_mysql_id.keys())
    pending = [c for c in categories if int(c["source_id"]) not in existing_source_ids]
    pending.sort(key=lambda c: (c.get("level") or 0, int(c["source_id"])))

    updated_map = dict(source_id_to_mysql_id)
    rows: list[dict[str, Any]] = []

    while pending:
        still_pending: list[dict[str, Any]] = []
        progress = False

        for cat in pending:
            source_id = int(cat["source_id"])
            source_parent_id = int(cat["source_parent_id"])

            if source_parent_id != 0 and source_parent_id not in updated_map:
                still_pending.append(cat)
                continue

            mysql_parent_id: int | None = None
            if source_parent_id != 0:
                mysql_parent_id = updated_map[source_parent_id]

            updated_map[source_id] = next_id
            rows.append(
                {
                    "id": next_id,
                    "name": cat["name"],
                    "description": cat.get("description"),
                    "level": cat.get("level"),
                    "parent_id": mysql_parent_id,
                    "source_id": source_id,
                    "source_parent_id": source_parent_id,
                }
            )
            next_id += 1
            progress = True

        if not progress and still_pending:
            for cat in still_pending:
                logger.warning(
                    "Parent source_id=%s not found for category source_id=%s, skip",
                    cat["source_parent_id"],
                    cat["source_id"],
                )
            break

        pending = still_pending

    return rows, updated_map, next_id


def _to_element_rows(
    elements: list[dict[str, Any]],
    source_id_to_mysql_id: dict[int, int],
) -> list[dict[str, Any]]:
    """将 ODPS 元素的 hive category_id 映射为 MySQL 分类 id。"""
    rows: list[dict[str, Any]] = []
    for row in elements:
        if row.get("name") is None or row.get("category_id") is None:
            continue

        source_category_id = int(row["category_id"])
        mysql_category_id = source_id_to_mysql_id.get(source_category_id)
        if mysql_category_id is None:
            logger.warning(
                "Category source_id=%s not found for element %s, skip",
                source_category_id,
                row["name"],
            )
            continue

        rows.append(
            {
                "name": str(row["name"]),
                "category_id": mysql_category_id,
                "source_category_id": source_category_id,
            }
        )

    return rows


def sync_global_tree_odps_to_mysql(partition_date: str | None = None) -> dict:
    """
    从 ODPS 拉取全局树元素与分类，写入 MySQL。

    Args:
        partition_date: 分区日期 (YYYYMMDD)，默认昨天
    """
    if partition_date is None:
        partition_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")

    logger.info("Starting global tree ODPS → MySQL sync for partition: %s", partition_date)

    odps = get_odps_client()
    raw_elements = odps.fetch_pattern_mining_elements(partition_date)
    raw_categories = odps.fetch_pattern_mining_categories(partition_date)
    logger.info(
        "Fetched %d elements and %d categories from ODPS",
        len(raw_elements),
        len(raw_categories),
    )

    category_by_id = {int(c["id"]): c for c in raw_categories if c.get("id") is not None}

    seed_ids = {
        int(row["category_id"])
        for row in raw_elements
        if row.get("category_id") is not None
    }
    categories = _collect_categories_with_ancestors(seed_ids, category_by_id)
    category_source_rows = _dedupe_category_source_rows(_to_category_source_rows(categories))

    logger.info(
        "Resolved %d elements and %d categories (with ancestors) for sync",
        len(raw_elements),
        len(category_source_rows),
    )

    with get_session() as session:
        element_repo = GlobalTreeElementRepository(session)
        category_repo = GlobalTreeCategoryRepository(session)

        source_id_to_mysql_id = category_repo.get_source_id_map()
        logger.info("Loaded %d existing categories from MySQL", len(source_id_to_mysql_id))

        existing_source_ids = set(source_id_to_mysql_id.keys())
        categories_already_exist = sum(
            1 for c in category_source_rows if int(c["source_id"]) in existing_source_ids
        )

        next_id = category_repo.get_max_id() + 1
        new_category_rows, _, _ = _prepare_new_category_rows(
            category_source_rows,
            source_id_to_mysql_id,
            next_id,
        )
        category_inserted = category_repo.bulk_insert_ignore(new_category_rows)

        # 插入后从 DB 重载映射，确保元素 category_id 与库内真实 id 一致
        source_id_to_mysql_id = category_repo.get_source_id_map()
        elements = _dedupe_element_rows(_to_element_rows(raw_elements, source_id_to_mysql_id))
        element_inserted = element_repo.bulk_insert_ignore(elements)

    categories_failed = len(category_source_rows) - categories_already_exist - len(new_category_rows)

    result = {
        "partition_date": partition_date,
        "elements_fetched": len(raw_elements),
        "elements_inserted": element_inserted,
        "elements_skipped": len(raw_elements) - len(elements),
        "categories_fetched": len(category_source_rows),
        "categories_inserted": category_inserted,
        "categories_already_exist": categories_already_exist,
        "categories_failed": categories_failed,
        "synced_at": datetime.now().isoformat(),
    }
    logger.info("Global tree sync completed: %s", result)
    return result


if __name__ == "__main__":
    import sys

    from supply_infra.scheduler.cli_result import run_cli

    run_cli(
        lambda: sync_global_tree_odps_to_mysql(
            partition_date=sys.argv[1] if len(sys.argv) > 1 else None,
        ),
        label="sync_global_tree_odps_to_mysql",
    )
