"""按行同步子表。

带 ID 的行原地修改，没有 ID 时按业务键对上尚未占用的原行，其余才新增。
本次提交里没有的活动行才软删除。不要整表软删后再插入，否则外键会指向已删除的行。
"""
from __future__ import annotations

from datetime import datetime
from typing import Any, Callable, Dict, List, Optional

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid


def _item_id(item: Dict[str, Any]) -> str:
    return str(item.get("id") or "").strip()


def _sort_key(row: Any):
    order_num = getattr(row, "order_num", None)
    create_time = getattr(row, "create_time", None)
    return (
        order_num if order_num is not None else 0,
        create_time or datetime.min,
        str(getattr(row, "id", "") or ""),
    )


def sync_child_rows(
    db: Session,
    model,
    *,
    parent_attr: str,
    parent_id: str,
    items: List[Dict[str, Any]],
    build_row: Callable[[str], Any],
    apply_fields: Callable[[Any, Dict[str, Any], int, datetime], None],
    tenant_no: Optional[str] = None,
    tenant_attr: str = "tenant_no",
    user_id: Optional[str] = None,
    now: Optional[datetime] = None,
    natural_key: Optional[Callable[[Dict[str, Any]], str]] = None,
    row_natural_key: Optional[Callable[[Any], str]] = None,
    duplicate_message: str = "编号重复",
    foreign_message: str = "记录不属于当前数据",
    extra_filters: Optional[Dict[str, Any]] = None,
    reuse_deleted: bool = False,
) -> List[Any]:
    now = now or datetime.now()
    payload = list(items or [])
    seen_ids = set()
    for item in payload:
        row_id = _item_id(item)
        if not row_id:
            continue
        if row_id in seen_ids:
            raise ValueError(duplicate_message)
        seen_ids.add(row_id)

    def _scoped(query):
        query = query.filter(getattr(model, parent_attr) == parent_id)
        if tenant_no is not None and hasattr(model, tenant_attr):
            query = query.filter(getattr(model, tenant_attr) == tenant_no)
        for key, value in (extra_filters or {}).items():
            query = query.filter(getattr(model, key) == value)
        return query

    active_rows = _scoped(db.query(model).filter(model.del_flag == 0)).all()
    active_by_id = {row.id: row for row in active_rows}

    for row_id in seen_ids:
        if row_id in active_by_id:
            continue
        other = db.query(model).filter(model.id == row_id).first()
        if other is None:
            continue
        other_parent = getattr(other, parent_attr)
        other_tenant = getattr(other, tenant_attr, None) if tenant_no is not None else tenant_no
        if other_parent != parent_id or (
            tenant_no is not None and (other_tenant or "") != tenant_no
        ):
            raise ValueError(foreign_message)

    queues: Dict[str, List[Any]] = {}
    deleted_queues: Dict[str, List[Any]] = {}
    if natural_key is not None and row_natural_key is not None:
        for row in sorted(active_rows, key=_sort_key):
            key = (row_natural_key(row) or "").strip()
            if not key:
                continue
            queues.setdefault(key, []).append(row)
        if reuse_deleted:
            deleted_rows = _scoped(db.query(model).filter(model.del_flag == 1)).all()
            for row in sorted(deleted_rows, key=_sort_key):
                key = (row_natural_key(row) or "").strip()
                if not key:
                    continue
                deleted_queues.setdefault(key, []).append(row)

    claimed = set()
    assigned: List[Optional[Any]] = [None] * len(payload)

    def _take(row: Any) -> Any:
        if row.id in claimed:
            raise ValueError(duplicate_message)
        claimed.add(row.id)
        return row

    for idx, item in enumerate(payload):
        row_id = _item_id(item)
        if not row_id:
            continue
        if row_id in active_by_id:
            assigned[idx] = _take(active_by_id[row_id])
            continue
        row = db.query(model).filter(model.id == row_id).first()
        if row is None:
            row = build_row(row_id)
            db.add(row)
        assigned[idx] = _take(row)

    for bucket in queues.values():
        bucket[:] = [row for row in bucket if row.id not in claimed]

    for idx, item in enumerate(payload):
        if assigned[idx] is not None:
            continue
        row = None
        if natural_key is not None:
            key = (natural_key(item) or "").strip()
            bucket = queues.get(key) if key else None
            while bucket:
                candidate = bucket.pop(0)
                if candidate.id not in claimed:
                    row = candidate
                    break
        if row is None and reuse_deleted and natural_key is not None:
            key = (natural_key(item) or "").strip()
            bucket = deleted_queues.get(key) if key else None
            while bucket:
                candidate = bucket.pop(0)
                if candidate.id not in claimed:
                    row = candidate
                    break
        if row is None:
            row = build_row(generate_uuid())
            db.add(row)
        assigned[idx] = _take(row)

    result: List[Any] = []
    for idx, item in enumerate(payload):
        row = assigned[idx]
        setattr(row, parent_attr, parent_id)
        if tenant_no is not None and hasattr(row, tenant_attr):
            setattr(row, tenant_attr, tenant_no)
        apply_fields(row, item, idx, now)
        row.del_flag = 0
        if hasattr(row, "update_by"):
            row.update_by = user_id
        if hasattr(row, "update_time"):
            row.update_time = now
        db.add(row)
        result.append(row)

    if reuse_deleted and row_natural_key is not None:
        keys_to_clear = set()
        for row in result:
            keys_to_clear.add((row_natural_key(row) or "").strip())
        for row in active_rows:
            if row.id not in claimed:
                keys_to_clear.add((row_natural_key(row) or "").strip())
        removed_tombstone = False
        for key in keys_to_clear:
            if not key:
                continue
            for tomb in deleted_queues.get(key, []):
                if tomb.id in claimed:
                    continue
                db.delete(tomb)
                removed_tombstone = True
        if removed_tombstone:
            db.flush()

    for row in active_rows:
        if row.id in claimed:
            continue
        row.del_flag = 1
        if hasattr(row, "update_by"):
            row.update_by = user_id
        if hasattr(row, "update_time"):
            row.update_time = now
        db.add(row)
    return result


def sync_link_rows(
    db: Session,
    model,
    *,
    parent_attr: str,
    parent_id: str,
    link_attr: str,
    link_ids: List[str],
    tenant_no: Optional[str] = None,
    user_id: Optional[str] = None,
    duplicate_message: str = "关联重复",
    foreign_message: str = "关联不属于当前数据",
) -> List[Any]:
    """按关联 id 更新关系行。重新加回已删除的关联时恢复原行，避免唯一索引冲突。"""
    now = datetime.now()
    ordered: List[Dict[str, Any]] = []
    seen = set()
    for raw in link_ids or []:
        link_id = str(raw or "").strip()
        if not link_id or link_id in seen:
            continue
        seen.add(link_id)
        ordered.append({link_attr: link_id})

    def build_row(row_id: str):
        data = {
            "id": row_id,
            parent_attr: parent_id,
            link_attr: "",
            "create_by": user_id,
            "create_dept_id": "",
            "create_time": now,
        }
        if tenant_no is not None and hasattr(model, "tenant_no"):
            data["tenant_no"] = tenant_no
        return model(**data)

    def apply_row(row: Any, item: Dict[str, Any], idx: int, _now: datetime) -> None:
        setattr(row, link_attr, item[link_attr])
        if tenant_no is not None and hasattr(row, "tenant_no"):
            row.tenant_no = tenant_no
        if hasattr(row, "order_num"):
            row.order_num = idx

    return sync_child_rows(
        db,
        model,
        parent_attr=parent_attr,
        parent_id=parent_id,
        tenant_no=tenant_no,
        items=ordered,
        user_id=user_id,
        now=now,
        build_row=build_row,
        apply_fields=apply_row,
        natural_key=lambda item: item[link_attr],
        row_natural_key=lambda row: str(getattr(row, link_attr) or "").strip(),
        duplicate_message=duplicate_message,
        foreign_message=foreign_message,
        reuse_deleted=True,
    )