import logging
from uuid import UUID

from geoalchemy2 import WKTElement
from sqlalchemy import and_, select, text
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from src.apps.plots.models.plot import Plot
from src.apps.plots.models.plot_group import PlotGroup, PlotGroupMember
from src.core.exceptions import NotFoundError, ValidationError
from src.core.utils.geometry import geometry_to_geojson

logger = logging.getLogger(__name__)

# Members within this many metres of each other count as adjacent (touching plots
# drawn as separate polygons rarely share exact vertices).
_ADJACENCY_TOLERANCE_M = 2.0


class PlotGroupService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def _load_group(self, group_id: UUID, tenant_id: UUID) -> PlotGroup | None:
        result = await self.db.execute(
            select(PlotGroup)
            .options(selectinload(PlotGroup.members))
            .where(and_(PlotGroup.id == group_id, PlotGroup.tenant_id == tenant_id))
        )
        return result.scalar_one_or_none()

    async def create(self, tenant_id: UUID, data: dict) -> dict:
        plot_ids: list[UUID] = data["plot_ids"]

        # All plots must belong to this tenant (BOLA / API1).
        result = await self.db.execute(
            select(Plot).where(
                and_(Plot.tenant_id == tenant_id, Plot.id.in_(plot_ids))
            )
        )
        plots = result.scalars().all()
        if len(plots) != len(plot_ids):
            raise NotFoundError("One or more plots were not found in this cemetery")

        section_id = plots[0].section_id
        with_geom = [p for p in plots if p.geometry is not None]

        # Adjacency validation only where geometries exist (image/demo mode has none).
        if len(with_geom) >= 2:
            await self._assert_adjacent(tenant_id, [p.id for p in with_geom])

        group = PlotGroup(
            tenant_id=tenant_id,
            section_id=section_id,
            group_type=data.get("group_type", "custom"),
            label=data.get("label"),
            capacity=data.get("capacity") or len(plot_ids),
        )

        # Group boundary = ST_Union of member geometries, only when it collapses to
        # a single polygon (adjacent plots); disjoint members leave boundary NULL.
        if len(with_geom) == len(plots) and with_geom:
            union_wkt = await self._union_polygon_wkt([p.id for p in with_geom])
            if union_wkt:
                group.boundary = WKTElement(union_wkt, srid=4326, extended=False)

        self.db.add(group)
        await self.db.flush()

        for pid in plot_ids:
            self.db.add(PlotGroupMember(plot_group_id=group.id, plot_id=pid))
        await self.db.flush()

        logger.info(
            "plot_group created tenant=%s group=%s members=%s type=%s",
            tenant_id, group.id, len(plot_ids), group.group_type,
        )
        return self._serialize(group, plot_ids)

    async def _assert_adjacent(self, tenant_id: UUID, ids: list[UUID]) -> None:
        """Every geometry-bearing member must touch/lie within tolerance of another.

        Rejects a selection where any plot is spatially isolated from the rest.
        """
        str_ids = [str(i) for i in ids]
        row = await self.db.execute(
            text(
                """
                SELECT count(DISTINCT pid) AS connected FROM (
                    SELECT a.id AS pid
                    FROM plots a JOIN plots b ON a.id <> b.id
                    WHERE a.tenant_id = :tid
                      AND a.id = ANY(:ids) AND b.id = ANY(:ids)
                      AND a.geometry IS NOT NULL AND b.geometry IS NOT NULL
                      AND ST_DWithin(a.geometry, b.geometry, :tol)
                ) t
                """
            ),
            {"tid": str(tenant_id), "ids": str_ids, "tol": _ADJACENCY_TOLERANCE_M},
        )
        connected = int(row.scalar_one() or 0)
        if connected < len(ids):
            raise ValidationError(
                "Selected plots must be adjacent to be grouped."
            )

    async def _union_polygon_wkt(self, ids: list[UUID]) -> str | None:
        str_ids = [str(i) for i in ids]
        row = await self.db.execute(
            text(
                """
                SELECT ST_AsText(ST_Union(geometry::geometry)) AS wkt
                FROM plots WHERE id = ANY(:ids) AND geometry IS NOT NULL
                """
            ),
            {"ids": str_ids},
        )
        wkt = row.scalar_one_or_none()
        if wkt and wkt.upper().startswith("POLYGON"):
            return wkt
        return None

    async def delete(self, group_id: UUID, tenant_id: UUID) -> None:
        group = await self._load_group(group_id, tenant_id)
        if not group:
            raise NotFoundError("Group not found")
        await self.db.delete(group)  # members cascade
        await self.db.flush()
        logger.info("plot_group deleted tenant=%s group=%s", tenant_id, group_id)

    def _serialize(self, group: PlotGroup, plot_ids: list[UUID]) -> dict:
        return {
            "id": group.id,
            "tenant_id": group.tenant_id,
            "section_id": group.section_id,
            "group_type": group.group_type,
            "label": group.label,
            "capacity": group.capacity,
            "plot_ids": plot_ids,
            "boundary": geometry_to_geojson(group.boundary),
        }