import asyncio
import logging
from datetime import datetime
import time
from typing import Literal

from django.db import connection
from django.db.utils import IntegrityError
from django.utils import timezone

from api_v1 import models as api_models
from flower.utils.mingdaoyun.fetch import (
    fetch_plate_orders_from_mingdaoyun,
    fetch_row_by_rowid_from_mingdaoyun,
)
from flower.utils.mingdaoyun.mappings import plate_order_related_worksheet_map
from flower.utils.mingdaoyun.relations import (
    extract_plate_order_related_rowids,
    flatten_plate_order_related_row,
)

logger = logging.getLogger(__name__)

_MDY_PLATE_ORDER_STAGING_PK_CONSTRAINT = "api_mdy_plate_order_staging_pkey"


def _reset_mdy_plate_order_staging_id_sequence() -> None:
    """修复 PostgreSQL 自增序列落后导致的主键重复（duplicate key violates ..._pkey）。

    常见场景：数据库 restore/copy 后，sequence 未随数据同步到最新值。
    """
    table_name = api_models.MDYPlateOrderStaging._meta.db_table  # 'api_mdy_plate_order_staging'
    pk_column = api_models.MDYPlateOrderStaging._meta.pk.column  # 'id'

    with connection.cursor() as cursor:
        cursor.execute(f"SELECT COALESCE(MAX({pk_column}), 1) FROM {table_name}")
        max_id = cursor.fetchone()[0] or 1

        cursor.execute("SELECT pg_get_serial_sequence(%s, %s)", [table_name, pk_column])
        seq_name = cursor.fetchone()[0]
        if not seq_name:
            logger.warning("未找到 %s.%s 的 serial sequence，跳过 setval", table_name, pk_column)
            return

        # is_called=true => nextval 会返回 max_id+1
        cursor.execute("SELECT setval(%s, %s, true)", [seq_name, max_id])
        logger.warning(
            "已重置 sequence=%s 到 max(id)=%s（下次插入将从 %s 开始）",
            seq_name,
            max_id,
            max_id + 1,
        )


def _is_pk_duplicate_error(exc: Exception) -> bool:
    """判断是否为主键约束冲突（多见于 sequence 未对齐）。"""
    msg = str(exc) or ""
    return _MDY_PLATE_ORDER_STAGING_PK_CONSTRAINT in msg


def _parse_mdy_datetime(value: str | None):
    if not value:
        return None
    try:
        dt = datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
    except ValueError:
        return None
    if timezone.is_naive(dt):
        dt = timezone.make_aware(dt, timezone.get_current_timezone())
    return dt


async def _fetch_related_flattened(
    plate_order_row: dict,
    *,
    max_related_per_type: int = 5,
    request_interval_seconds: float = 0.0,
) -> list[dict]:
    """根据开版主表 row 的 Relation 字段，拉取并碾平关联表数据。"""

    related_rowids_map = extract_plate_order_related_rowids(plate_order_row)
    if not related_rowids_map:
        return []

    flattened: list[dict] = []
    for related_key, rowids in related_rowids_map.items():
        wsid = plate_order_related_worksheet_map.get(related_key)
        if not wsid:
            continue

        for rid in rowids[: max_related_per_type or 0]:
            related_row = await fetch_row_by_rowid_from_mingdaoyun(
                worksheet_id=wsid,
                rowid=rid,
            )
            if not related_row:
                continue

            flattened.append(flatten_plate_order_related_row(related_key, related_row))
            if request_interval_seconds:
                await asyncio.sleep(request_interval_seconds)

    return flattened


async def _fetch_related_for_rows(
    rows: list[dict],
    *,
    max_related_per_type: int,
    request_interval_seconds: float = 0.0,
) -> dict[str, list[dict]]:
    """批量拉取并碾平关联表数据（仅网络请求，不涉及 ORM）。"""

    result: dict[str, list[dict]] = {}
    for row in rows:
        rowid = row.get("rowid")
        if not rowid:
            continue
        result[rowid] = await _fetch_related_flattened(
            row,
            max_related_per_type=max_related_per_type,
            request_interval_seconds=request_interval_seconds,
        )
    return result


def sync_mdy_plate_orders_to_staging(
    *,
    page_size: int = 100,
    max_pages: int | None = None,
    max_records: int | None = None,
    with_related: bool = True,
    max_related_per_type: int = 5,
    request_interval_seconds: float = 0.02,
    use_checkpoint: bool = True,
    update_checkpoint: bool = True,
    sort_direction: Literal["asc", "desc"] = "asc",
) -> dict:
    """同步明道云开版数据表到暂存表（含可选跨表关联数据）。

    - 写入 `api_v1.MDYPlateOrderStaging`：
      - `raw`：整行原始数据
      - `related`：跨表数据（碾平 list[dict]）
    - 进度记录到 `api_v1.DataSync`（table_name=plate_order）

    说明：
    - max_pages 表示“单次任务最多处理多少页”（不是最大页码）
    - max_records 表示“单次任务最多处理多少条记录”（0/None 表示不限制）
    - use_checkpoint 控制本次是否读取 DataSync 游标
    - update_checkpoint 控制本次是否写入 DataSync 记录
    - sort_direction 为 `asc`（默认）或 `desc`，决定按时间升/降序抓取
    """

    max_records = max_records or 0
    sort_direction = (sort_direction or "asc").lower()
    if sort_direction not in {"asc", "desc"}:
        raise ValueError("sort_direction must be 'asc' or 'desc'")
    is_asc = sort_direction == "asc"

    last_sync = None
    if use_checkpoint:
        last_sync = (
            api_models.DataSync.objects.filter(
                table_name=api_models.DataSync.TableName.PLATE_ORDER
            )
            .order_by("-created_at")
            .first()
        )
    last_ctime = last_sync.last_ctime if last_sync else None
    last_rowid = last_sync.last_rowid if last_sync else ""
    start_page_index = last_sync.page_index if last_sync and last_sync.page_index else 1

    synced_rows = 0
    page_index = max(1, start_page_index)
    pages_processed = 0
    total_count = 0
    latest_ctime = last_ctime
    latest_rowid = last_rowid

    while True:
        if max_pages is not None and pages_processed >= max_pages:
            break
        if max_records and synced_rows >= max_records:
            break

        rows, total = asyncio.run(
            fetch_plate_orders_from_mingdaoyun(
                page=page_index,
                page_size=page_size,
                sort_id="ctime",
                is_asc=is_asc,
            )
        )
        if request_interval_seconds:
            time.sleep(request_interval_seconds)
        total_count = total
        if not rows:
            break

        hit_max_records = False
        to_process: list[dict] = []
        for row in rows:
            if max_records and synced_rows + len(to_process) >= max_records:
                hit_max_records = True
                break

            rowid = row.get("rowid")
            if not rowid:
                continue

            record_ctime = _parse_mdy_datetime(row.get("ctime"))

            # 跳过已同步到的游标
            if use_checkpoint and last_ctime and record_ctime:
                if record_ctime < last_ctime:
                    continue
                if record_ctime == last_ctime and last_rowid and rowid == last_rowid:
                    continue

            to_process.append(row)

        related_map: dict[str, list[dict]] = {}
        if with_related and to_process:
            related_map = asyncio.run(
                _fetch_related_for_rows(
                    to_process,
                    max_related_per_type=max_related_per_type,
                    request_interval_seconds=request_interval_seconds,
                )
            )

        for row in to_process:
            rowid = row.get("rowid")
            if not rowid:
                continue

            defaults = {
                "ctime": _parse_mdy_datetime(row.get("ctime")),
                "utime": _parse_mdy_datetime(row.get("utime")),
                "raw": row,
                "related": related_map.get(rowid, []),
            }

            try:
                api_models.MDYPlateOrderStaging.objects.update_or_create(
                    mdy_rowid=rowid,
                    defaults=defaults,
                )
            except IntegrityError as exc:
                # 仅对“主键重复（通常为 sequence 未对齐）”做一次自愈重试。
                if _is_pk_duplicate_error(exc):
                    logger.warning(
                        "检测到暂存表主键冲突，尝试重置序列后重试一次：rowid=%s err=%s",
                        rowid,
                        exc,
                    )
                    _reset_mdy_plate_order_staging_id_sequence()
                    api_models.MDYPlateOrderStaging.objects.update_or_create(
                        mdy_rowid=rowid,
                        defaults=defaults,
                    )
                else:
                    raise
            synced_rows += 1

            record_ctime = _parse_mdy_datetime(row.get("ctime"))
            if record_ctime:
                if latest_ctime is None:
                    latest_ctime = record_ctime
                    latest_rowid = rowid
                else:
                    should_update = (
                        record_ctime > latest_ctime if is_asc else record_ctime < latest_ctime
                    )
                    if should_update:
                        latest_ctime = record_ctime
                        latest_rowid = rowid
                    elif record_ctime == latest_ctime:
                        latest_rowid = rowid

        pages_processed += 1
        if hit_max_records:
            break

        if len(rows) < page_size:
            break
        page_index += 1

    record_last_ctime = latest_ctime or last_ctime
    record_last_rowid = latest_rowid or last_rowid

    if update_checkpoint:
        api_models.DataSync.objects.create(
            table_name=api_models.DataSync.TableName.PLATE_ORDER,
            page_index=page_index,
            page_size=page_size,
            synced_rows=synced_rows,
            total_count=total_count,
            last_ctime=record_last_ctime,
            last_rowid=record_last_rowid,
            note=(
                f"{sort_direction} scan; with_related={with_related}; "
                f"use_checkpoint={use_checkpoint}"
            ),
        )

    payload = {
        "synced_rows": synced_rows,
        "page_index": page_index,
        "page_size": page_size,
        "total_count": total_count,
        "last_ctime": record_last_ctime.isoformat() if record_last_ctime else None,
        "with_related": with_related,
        "sort_direction": sort_direction,
        "use_checkpoint": use_checkpoint,
        "update_checkpoint": update_checkpoint,
    }
    logger.info("明道云开版暂存同步完成: %s", payload)
    return payload