import logging from typing import Optional from uuid import UUID from geoalchemy2 import WKTElement from sqlalchemy import and_, func, or_, 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.services.plot_status_service import PlotStatusService from src.apps.records.models.record import Record from src.apps.sections.models.section import Section from src.core.constants import PlotStatus from src.core.exceptions import ConflictError, NotFoundError, ValidationError from src.core.utils.geometry import geometry_to_geojson, validate_polygon_geojson logger = logging.getLogger(__name__) # Per-tenant soft ceiling for the single-call map bootstrap (PRD performance note / # security A04). Above this we log a warning; we never silently truncate. _MAP_PLOT_WARN_THRESHOLD = 2000 class PlotService: def __init__(self, db: AsyncSession): self.db = db def _search_conditions(self, search: str): """Match plot_ref / label_manual OR the linked record's name (AC-04).""" like = f"%{search}%" return or_( Plot.plot_ref.ilike(like), Plot.label_manual.ilike(like), Record.first_name.ilike(like), Record.last_name.ilike(like), Record.maiden_name.ilike(like), ) async def list_plots( self, tenant_id: UUID, skip: int = 0, limit: int = 50, section_id: Optional[UUID] = None, status: Optional[str] = None, search: Optional[str] = None, ) -> tuple[list, int]: conditions = [Plot.tenant_id == tenant_id] if section_id is not None: conditions.append(Plot.section_id == section_id) if status is not None: conditions.append(Plot.status == status) # Left-join the (non-deleted) linked record so search can match names too. record_join = and_(Record.plot_id == Plot.id, Record.deleted_at.is_(None)) if search: conditions.append(self._search_conditions(search)) count_stmt = select(func.count(func.distinct(Plot.id))).select_from(Plot) data_stmt = select(Plot) if search: count_stmt = count_stmt.outerjoin(Record, record_join) data_stmt = data_stmt.outerjoin(Record, record_join) total = (await self.db.execute(count_stmt.where(and_(*conditions)))).scalar_one() result = await self.db.execute( data_stmt.options( selectinload(Plot.section), selectinload(Plot.plot_type), ) .where(and_(*conditions)) .order_by(Plot.plot_ref.asc()) .offset(skip) .limit(limit) ) return result.scalars().all(), total async def get_by_id(self, plot_id: UUID, tenant_id: UUID) -> Optional[Plot]: result = await self.db.execute( select(Plot) .options( selectinload(Plot.section), selectinload(Plot.plot_type), ) .where(and_(Plot.id == plot_id, Plot.tenant_id == tenant_id)) ) return result.scalar_one_or_none() async def create(self, tenant_id: UUID, data: dict) -> Plot: existing = await self.db.execute( select(Plot).where( and_( Plot.tenant_id == tenant_id, Plot.plot_ref == data["plot_ref"], ) ) ) if existing.scalar_one_or_none(): raise ConflictError( f"A plot with reference '{data['plot_ref']}' already exists in this cemetery" ) plot = Plot( tenant_id=tenant_id, plot_ref=data["plot_ref"], section_id=data.get("section_id"), plot_type_id=data.get("plot_type_id"), status=data.get("status", PlotStatus.VACANT.value), latitude=data.get("latitude"), longitude=data.get("longitude"), price_override=data.get("price_override"), notes=data.get("notes"), public_description=data.get("public_description"), is_veteran_section=data.get("is_veteran_section", False), label_manual=data.get("label_manual"), ) # Mirror manually-entered lat/lng into centroid (image-mode tenants). if plot.latitude is not None and plot.longitude is not None: plot.centroid = WKTElement( f"POINT({plot.longitude} {plot.latitude})", srid=4326, extended=False ) self.db.add(plot) await self.db.flush() return await self.get_by_id(plot.id, tenant_id) async def update(self, plot_id: UUID, tenant_id: UUID, data: dict) -> Plot: plot = await self.get_by_id(plot_id, tenant_id) if not plot: raise NotFoundError("Plot not found") # Geometry write path: validate → store → recompute centroid → mirror lat/lng. geometry = data.pop("geometry", None) if geometry is not None: poly = validate_polygon_geojson(geometry) # 422 on malformed input plot.geometry = WKTElement(poly.wkt, srid=4326, extended=False) cx, cy = poly.centroid.x, poly.centroid.y # lng, lat plot.centroid = WKTElement( f"POINT({cx} {cy})", srid=4326, extended=False ) plot.latitude = round(cy, 7) plot.longitude = round(cx, 7) # Whitelisted scalar fields only (tenant_id/section computed fields excluded — API3). allowed = { "plot_ref", "section_id", "plot_type_id", "status", "latitude", "longitude", "price_override", "notes", "public_description", "is_veteran_section", "rotation_deg", "reserved_by", "reserved_at", "label_manual", } # Status, when edited via the form, must be a valid enum value (same guard # as the dedicated change_status endpoint). new_status = data.get("status") if new_status is not None: valid_statuses = {s.value for s in PlotStatus} if new_status not in valid_statuses: raise ValidationError( f"Invalid status '{new_status}'. Must be one of: " f"{', '.join(sorted(valid_statuses))}" ) # plot_ref edits must stay unique per tenant (AC-11). new_ref = data.get("plot_ref") if new_ref and new_ref != plot.plot_ref: clash = await self.db.execute( select(Plot).where( and_( Plot.tenant_id == tenant_id, Plot.plot_ref == new_ref, Plot.id != plot_id, ) ) ) if clash.scalar_one_or_none(): raise ConflictError( f"A plot with reference '{new_ref}' already exists in this cemetery" ) for field, value in data.items(): if field in allowed: setattr(plot, field, value) # If lat/lng were changed directly (no geometry), keep centroid in sync. if geometry is None and ("latitude" in data or "longitude" in data): if plot.latitude is not None and plot.longitude is not None: plot.centroid = WKTElement( f"POINT({plot.longitude} {plot.latitude})", srid=4326, extended=False, ) await self.db.flush() return await self.get_by_id(plot_id, tenant_id) async def change_status(self, plot_id: UUID, tenant_id: UUID, status: str) -> Plot: plot = await self.get_by_id(plot_id, tenant_id) if not plot: raise NotFoundError("Plot not found") valid_statuses = {s.value for s in PlotStatus} if status not in valid_statuses: raise ValidationError( f"Invalid status '{status}'. Must be one of: {', '.join(sorted(valid_statuses))}" ) plot.status = status await self.db.flush() return await self.get_by_id(plot_id, tenant_id) async def revert_to_default(self, plot_id: UUID, tenant_id: UUID) -> tuple[Plot, str]: """Revert an occupied/reserved plot back to the tenant's default (vacant) status and clear its reservation fields (AC-18 / API1 / SEC-02/SEC-11). Tenant isolation is enforced by get_by_id (id AND tenant_id); a cross-tenant plot id yields NotFound → HTTP 404, never a cross-tenant mutation. Returns (plot, prior_status) so the caller can emit the audit log. """ plot = await self.get_by_id(plot_id, tenant_id) if not plot: raise NotFoundError("Plot not found") prior_status = plot.status # Resolve the tenant's default "vacant" catalog entry; fall back to the # canonical enum value if the catalog was customised. statuses = await PlotStatusService(self.db).list_or_seed(tenant_id) default = next( (s for s in statuses if s.status_key == PlotStatus.VACANT.value), next((s for s in statuses if s.is_default and s.status_key), None), ) plot.status = default.status_key if default else PlotStatus.VACANT.value plot.reserved_by = None plot.reserved_at = None await self.db.flush() logger.info( "plots.revert actor_tenant=%s plot_id=%s prior_status=%s new_status=%s", tenant_id, plot_id, prior_status, plot.status, ) return await self.get_by_id(plot_id, tenant_id), prior_status async def delete(self, plot_id: UUID, tenant_id: UUID) -> None: plot = await self.get_by_id(plot_id, tenant_id) if not plot: raise NotFoundError("Plot not found") # Blocked (409) while a burial record is linked — matches the "must be vacant" # guard and prevents orphaned records (AC-26 / SAC-04 / A04). linked = await self.db.execute( select(func.count(Record.id)).where( and_( Record.plot_id == plot_id, Record.tenant_id == tenant_id, Record.deleted_at.is_(None), ) ) ) if int(linked.scalar_one()) > 0: raise ConflictError( "This plot has a linked record and cannot be deleted. " "Move or remove the record first." ) if plot.status != PlotStatus.VACANT.value: raise ConflictError( "Only vacant plots can be deleted. " f"This plot has status '{plot.status}'." ) await self.db.delete(plot) await self.db.flush() # ── Map bootstrap (INDL-41, AC-01/02/03/17) ───────────────────────────────── async def _account_map_context(self, tenant_id: UUID) -> dict: row = (await self.db.execute( text( "SELECT map_mode, " "ST_Y(centroid::geometry) AS lat, ST_X(centroid::geometry) AS lng " "FROM accounts WHERE id = :aid" ), {"aid": str(tenant_id)}, )).first() map_mode = row[0] if row and row[0] else "image" center = None if row and row[1] is not None: center = [float(row[1]), float(row[2])] return {"map_mode": map_mode, "center": center} def _plot_feature(self, plot: Plot, color_map: dict) -> dict: geom = geometry_to_geojson(plot.geometry) if geom is None: geom = geometry_to_geojson(plot.centroid) if geom is None and plot.latitude is not None and plot.longitude is not None: geom = {"type": "Point", "coordinates": [float(plot.longitude), float(plot.latitude)]} record = plot.record if plot.record and plot.record.deleted_at is None else None occupant = None if record is not None: occupant = f"{record.first_name} {record.last_name}".strip() return { "type": "Feature", "geometry": geom, "properties": { "id": str(plot.id), "plot_ref": plot.plot_ref, "label": plot.label_manual or plot.plot_ref, "status": plot.status, "color": color_map.get(plot.status, "#94a3b8"), "section_id": str(plot.section_id) if plot.section_id else None, "section_code": plot.section.code if plot.section else None, "plot_type_name": plot.plot_type.name if plot.plot_type else None, "price": float(plot.price_override) if plot.price_override is not None else None, "reserved_by": plot.reserved_by, "reserved_at": plot.reserved_at.isoformat() if plot.reserved_at else None, "rotation_deg": float(plot.rotation_deg) if plot.rotation_deg is not None else None, "latitude": float(plot.latitude) if plot.latitude is not None else None, "longitude": float(plot.longitude) if plot.longitude is not None else None, "has_record": record is not None, "occupant_name": occupant, "record_id": str(record.id) if record is not None else None, }, } async def list_for_map(self, tenant_id: UUID) -> dict: ctx = await self._account_map_context(tenant_id) statuses = await PlotStatusService(self.db).list_or_seed(tenant_id) color_map = {s.status_key: s.color_hex for s in statuses if s.status_key} # Sections (with boundary + occupancy counts for the heat-map). sections_result = await self.db.execute( select(Section).where(Section.tenant_id == tenant_id) ) sections = sections_result.scalars().all() counts = await self._section_status_counts(tenant_id) section_features = [] for s in sections: c = counts.get(s.id, {"total": 0, "available": 0, "occupied": 0, "reserved": 0}) occupied_reserved = c["occupied"] + c["reserved"] occupancy_pct = ( round(occupied_reserved / c["total"] * 100, 1) if c["total"] else 0.0 ) section_features.append({ "type": "Feature", "geometry": geometry_to_geojson(s.boundary), "properties": { "id": str(s.id), "code": s.code, "name": s.name, "display_color": s.display_color, "total": c["total"], "available": c["available"], "occupancy_pct": occupancy_pct, "area_m2": float(s.area_m2) if s.area_m2 is not None else None, }, }) # Plots (with record for occupant/search). plots_result = await self.db.execute( select(Plot) .options( selectinload(Plot.section), selectinload(Plot.plot_type), selectinload(Plot.record), ) .where(Plot.tenant_id == tenant_id) .order_by(Plot.plot_ref.asc()) ) plots = plots_result.scalars().all() if len(plots) > _MAP_PLOT_WARN_THRESHOLD: logger.warning( "plots/map returning %s plots for tenant=%s (above %s soft ceiling)", len(plots), tenant_id, _MAP_PLOT_WARN_THRESHOLD, ) plot_features = [self._plot_feature(p, color_map) for p in plots] return { "map_mode": ctx["map_mode"], "center": ctx["center"], "statuses": [ { "id": str(s.id), "name": s.name, "color_hex": s.color_hex, "status_key": s.status_key, "is_default": s.is_default, "sort_order": s.sort_order, } for s in statuses ], "sections": {"type": "FeatureCollection", "features": section_features}, "plots": {"type": "FeatureCollection", "features": plot_features}, } async def _section_status_counts(self, tenant_id: UUID) -> dict: """One grouped query → {section_id: {total, available, occupied, reserved}}.""" result = await self.db.execute( select(Plot.section_id, Plot.status, func.count(Plot.id)) .where(Plot.tenant_id == tenant_id) .group_by(Plot.section_id, Plot.status) ) out: dict = {} for section_id, status, cnt in result.all(): bucket = out.setdefault( section_id, {"total": 0, "available": 0, "occupied": 0, "reserved": 0} ) cnt = int(cnt) bucket["total"] += cnt if status == PlotStatus.VACANT.value: bucket["available"] += cnt elif status == PlotStatus.OCCUPIED.value: bucket["occupied"] += cnt elif status == PlotStatus.RESERVED.value: bucket["reserved"] += cnt return out async def section_counts(self, tenant_id: UUID) -> dict: """Public helper for the sections list (available_count/total_count).""" return await self._section_status_counts(tenant_id)