"""Publish service — snapshot creation and house validation for version management.""" import math from typing import Any from sqlalchemy.orm import Session from app.models.floor import Floor from app.models.hotspot import Hotspot from app.models.house import House from app.models.panorama import Panorama from app.models.room import Room def create_snapshot(house_id: str, db: Session) -> dict[str, Any]: """Serialize the full state of a house into a dict for version snapshots. Returns a nested structure:: { "house": {...}, "floors": [ { "id": str, "name": str, "level": int, "sort_order": int, "rooms": [ { "id": str, "name": str, "type": str, "panoramas": [ { "id": str, "image_url": str, ... "hotspots": [...] } ] } ] } ] } Raises: ValueError: if the house does not exist. """ house = db.query(House).filter(House.id == house_id).first() if not house: raise ValueError("房屋不存在") floors_data: list[dict[str, Any]] = [] for floor in house.floors: rooms_data: list[dict[str, Any]] = [] for room in floor.rooms: panoramas_data: list[dict[str, Any]] = [] for pano in room.panoramas: hotspots_from = ( db.query(Hotspot) .filter(Hotspot.from_pano_id == pano.id) .all() ) panoramas_data.append( { "id": pano.id, "image_url": pano.image_url, "thumbnail_url": pano.thumbnail_url, "sequence": pano.sequence, "default_yaw": pano.default_yaw, "default_pitch": pano.default_pitch, "capture_order": pano.capture_order, "position_x": pano.position_x, "position_y": pano.position_y, "position_z": pano.position_z, "metadata": pano.metadata, "hotspots": [ { "id": h.id, "from_pano_id": h.from_pano_id, "to_pano_id": h.to_pano_id, "yaw": h.yaw, "pitch": h.pitch, "type": h.type, "label": h.label, "confidence": h.confidence, "status": h.status, "created_by": h.created_by, "metadata": h.metadata, } for h in hotspots_from ], } ) rooms_data.append( { "id": room.id, "name": room.name, "type": room.type, "sort_order": room.sort_order, "cover_pano_id": room.cover_pano_id, "panoramas": panoramas_data, } ) floors_data.append( { "id": floor.id, "name": floor.name, "level": floor.level, "sort_order": floor.sort_order, "rooms": rooms_data, } ) return { "house": { "id": house.id, "title": house.title, "address": house.address, "slug": house.slug, "cover_image_url": house.cover_image_url, "status": house.status, }, "floors": floors_data, } def validate_house(house_id: str, db: Session) -> list[dict[str, Any]]: """Validate a house for publishing readiness. Checks performed: - Orphan hotspot connections (to_pano_id does not exist) - One-way connections (no reverse hotspot) - Low confidence hotspots (< 0.7) - Reverse yaw mismatch (> 20 degrees) - Unnamed rooms - Rooms with no entry hotspot Returns: A list of issue dicts, each with ``type``, ``message``, ``severity``, and contextual identifiers. """ issues: list[dict[str, Any]] = [] house = db.query(House).filter(House.id == house_id).first() if not house: return [ { "type": "house_not_found", "message": "房屋不存在", "severity": "error", } ] all_pano_ids: set[str] = set() pano_to_room: dict[str, str] = {} room_has_entry: dict[str, bool] = {} room_names: dict[str, str] = {} for floor in house.floors: for room in floor.rooms: room_has_entry[room.id] = False room_names[room.id] = room.name or "" # Check unnamed rooms if not room.name or not room.name.strip(): issues.append( { "type": "unnamed_room", "message": f"楼层 '{floor.name}' 中存在未命名的房间", "severity": "warning", "room_id": room.id, } ) for pano in room.panoramas: all_pano_ids.add(pano.id) pano_to_room[pano.id] = room.id # Collect all hotspots belonging to panoramas in this house all_hotspots = ( db.query(Hotspot) .filter(Hotspot.from_pano_id.in_(all_pano_ids)) .all() if all_pano_ids else [] ) hotspot_pairs: dict[tuple[str, str], list[Hotspot]] = {} for h in all_hotspots: # Orphan destination if h.to_pano_id and h.to_pano_id not in all_pano_ids: issues.append( { "type": "orphan_hotspot", "message": f"热点 {h.id} 连接到不存在的全景图 {h.to_pano_id}", "severity": "error", "hotspot_id": h.id, } ) # Low confidence if h.confidence < 0.7: issues.append( { "type": "low_confidence", "message": f"热点 {h.id} 置信度过低 ({h.confidence})", "severity": "warning", "hotspot_id": h.id, } ) if h.to_pano_id: pair = (h.from_pano_id, h.to_pano_id) hotspot_pairs.setdefault(pair, []).append(h) # One-way connection and reverse-yaw checks for (frm, to), forward_hotspots in hotspot_pairs.items(): reverse_pair = (to, frm) if reverse_pair not in hotspot_pairs: issues.append( { "type": "one_way_connection", "message": ( f"存在单向连接: 全景图 {frm[:8]}... → {to[:8]}..." ), "severity": "warning", "from_pano_id": frm, "to_pano_id": to, } ) else: # Reverse yaw mismatch check reverse_hotspots = hotspot_pairs[reverse_pair] for fh in forward_hotspots: for rh in reverse_hotspots: yaw_diff = abs(fh.yaw - rh.yaw) yaw_diff = min(yaw_diff, 360 - yaw_diff) if yaw_diff > 20: issues.append( { "type": "reverse_yaw_mismatch", "message": ( f"双向热点偏航角不匹配，差值 " f"{yaw_diff:.1f}° (>20°)" ), "severity": "warning", "hotspot_ids": [fh.id, rh.id], } ) # Room entry hotspot check for h in all_hotspots: if h.type == "room_enter" and h.to_pano_id: to_room_id = pano_to_room.get(h.to_pano_id) if to_room_id: room_has_entry[to_room_id] = True for room_id, has_entry in room_has_entry.items(): if not has_entry and room_names.get(room_id): # Only flag if the room actually has panoramas room_panos = [ pid for pid, rid in pano_to_room.items() if rid == room_id ] if room_panos: issues.append( { "type": "no_entry_hotspot", "message": ( f"房间 '{room_names[room_id]}' 没有入口热点" ), "severity": "warning", "room_id": room_id, } ) return issues