"""Buildings and levels: the drawings a marker can be placed on (ADR-017). Reads are UNAUTHENTICATED. The printer installer map runs before anyone logs in and needs a blueprint to draw, exactly as the printer install-list and the slide feed already do. A level name, a blueprint path and a pixel size are not secrets - the marker positions drawn on them already render on kiosk pages. Writes require admin: adding a level changes where every marker on it appears. """ import os from flask import Blueprint, request, current_app from flask_jwt_extended import jwt_required from werkzeug.utils import secure_filename from shopdb.extensions import db from shopdb.core.models import Building, MapLevel, Asset, AuditLog from shopdb.utils.responses import success_response, error_response, ErrorCodes from shopdb.utils.authz import require_role from shopdb.api import send_upload from shopdb.utils.imagesize import image_size maplevels_bp = Blueprint('maplevels', __name__) # Same set the map blueprint upload already accepts. SVG stays allowed because a # floor plan is vector by nature; it is served through send_upload, which sends # the sandbox headers that stop one executing as script. BLUEPRINT_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg'} def _blueprint_dir(): return os.path.join(current_app.instance_path, 'maps') # ============================================================================= # Read - public # ============================================================================= @maplevels_bp.route('', methods=['GET']) def list_levels(): """Every active level, grouped by building, in display order. One call, because every consumer needs the whole list: the map switches between levels, and a hover preview has to resolve an arbitrary asset's level to a blueprint without knowing in advance which one it is. """ buildings = (Building.query.filter_by(isactive=True) .order_by(Building.sortorder, Building.buildingid).all()) # How many markers sit on each level, in one grouped query rather than one # per level. The admin page needs it to say what a rename or a resize is # about to affect, and it is what makes the delete refusal predictable # instead of a surprise. counts = dict(db.session.query(Asset.levelid, db.func.count(Asset.assetid)) .filter(Asset.levelid.isnot(None), Asset.isactive.is_(True)) .group_by(Asset.levelid).all()) payload = [] for building in buildings: entry = building.to_dict() for level in entry.get('levels', []): level['assetcount'] = counts.get(level['levelid'], 0) payload.append(entry) default = MapLevel.default_level() return success_response({ 'buildings': payload, 'defaultlevelid': default.levelid if default else None, 'totalplaced': sum(counts.values()), }) @maplevels_bp.route('/', methods=['GET']) def get_level(levelid): level = db.session.get(MapLevel, levelid) if not level or not level.isactive: return error_response(ErrorCodes.NOT_FOUND, 'No such level', http_code=404) return success_response(level.to_dict()) @maplevels_bp.route('//blueprint/', methods=['GET']) def serve_blueprint(filename, levelid=None): """Serve a level's blueprint image. Public, for the same reason as above.""" return send_upload(_blueprint_dir(), filename) # ============================================================================= # Write - admin # ============================================================================= @maplevels_bp.route('/buildings', methods=['POST']) @jwt_required() @require_role('admin') def create_building(): data = request.get_json() or {} name = (data.get('buildingname') or '').strip() if not name: return error_response(ErrorCodes.VALIDATION_ERROR, 'buildingname is required') if Building.query.filter(db.func.lower(Building.buildingname) == name.lower()).first(): return error_response(ErrorCodes.CONFLICT, f"Building '{name}' already exists", http_code=409) building = Building(buildingname=name, sortorder=int(data.get('sortorder') or 0)) db.session.add(building) db.session.commit() AuditLog.log('created', 'Building', entityid=building.buildingid, entityname=name) db.session.commit() return success_response(building.to_dict(), message='Building created', http_code=201) @maplevels_bp.route('/buildings/', methods=['PUT', 'PATCH']) @jwt_required() @require_role('admin') def update_building(buildingid): """Rename or reorder a building. Levels move with it; nothing repositions.""" building = db.session.get(Building, buildingid) if not building: return error_response(ErrorCodes.NOT_FOUND, 'No such building', http_code=404) data = request.get_json() or {} if 'buildingname' in data: name = (data['buildingname'] or '').strip() if not name: return error_response(ErrorCodes.VALIDATION_ERROR, 'buildingname cannot be blank') clash = Building.query.filter( db.func.lower(Building.buildingname) == name.lower(), Building.buildingid != buildingid).first() if clash: return error_response(ErrorCodes.CONFLICT, f"Building '{name}' already exists", http_code=409) building.buildingname = name if data.get('sortorder') is not None: building.sortorder = int(data['sortorder']) if 'isactive' in data: building.isactive = bool(data['isactive']) db.session.commit() AuditLog.log('updated', 'Building', entityid=buildingid, entityname=building.buildingname) db.session.commit() return success_response(building.to_dict(), message='Building updated') @maplevels_bp.route('', methods=['POST']) @jwt_required() @require_role('admin') def create_level(): data = request.get_json() or {} name = (data.get('levelname') or '').strip() buildingid = data.get('buildingid') if not name or not buildingid: return error_response(ErrorCodes.VALIDATION_ERROR, 'buildingname and buildingid are required') if not db.session.get(Building, buildingid): return error_response(ErrorCodes.NOT_FOUND, 'No such building', http_code=404) level = MapLevel( buildingid=buildingid, levelname=name, sortorder=int(data.get('sortorder') or 0), blueprintlight=data.get('blueprintlight'), blueprintdark=data.get('blueprintdark'), mapwidth=int(data.get('mapwidth') or 3300), mapheight=int(data.get('mapheight') or 2550), ) db.session.add(level) db.session.flush() _apply_default(level, data.get('isdefault')) db.session.commit() AuditLog.log('created', 'MapLevel', entityid=level.levelid, entityname=name) db.session.commit() return success_response(level.to_dict(), message='Level created', http_code=201) @maplevels_bp.route('/', methods=['PUT', 'PATCH']) @jwt_required() @require_role('admin') def update_level(levelid): level = db.session.get(MapLevel, levelid) if not level: return error_response(ErrorCodes.NOT_FOUND, 'No such level', http_code=404) data = request.get_json() or {} # The dimensions are the coordinate space every marker on this level is # expressed in, so changing them moves every marker relative to the image. # Report it rather than doing it silently; the caller decides whether to run # a transform (POST /api/assets/map/transform). warnings = [] for field in ('mapwidth', 'mapheight'): if field in data and int(data[field] or 0) != getattr(level, field): warnings.append( '%s changed from %s to %s; existing positions on this level are ' 'still in the old coordinate space' % (field, getattr(level, field), data[field])) for field in ('levelname', 'blueprintlight', 'blueprintdark'): if field in data: setattr(level, field, data[field]) for field in ('sortorder', 'mapwidth', 'mapheight'): if field in data and data[field] is not None: setattr(level, field, int(data[field])) if 'isactive' in data: level.isactive = bool(data['isactive']) _apply_default(level, data.get('isdefault')) db.session.commit() AuditLog.log('updated', 'MapLevel', entityid=level.levelid, entityname=level.levelname) db.session.commit() payload = level.to_dict() if warnings: payload['warnings'] = warnings return success_response(payload, message='Level updated') @maplevels_bp.route('/', methods=['DELETE']) @jwt_required() @require_role('admin') def delete_level(levelid): """Deactivate a level, refusing while assets are still placed on it. Deleting the drawing out from under a marker would leave a position in a coordinate space that no longer exists - unrenderable, and indistinguishable from a marker that was never placed. """ level = db.session.get(MapLevel, levelid) if not level: return error_response(ErrorCodes.NOT_FOUND, 'No such level', http_code=404) placed = Asset.query.filter_by(levelid=levelid, isactive=True).count() if placed: return error_response( ErrorCodes.CONFLICT, f'{placed} asset(s) are placed on this level. Move them to another ' f'level first.', http_code=409) if level.isdefault: return error_response( ErrorCodes.VALIDATION_ERROR, 'This is the default level. Make another level the default first.') level.isactive = False db.session.commit() AuditLog.log('deleted', 'MapLevel', entityid=levelid, entityname=level.levelname) db.session.commit() return success_response(message='Level deactivated') @maplevels_bp.route('//blueprint', methods=['POST']) @jwt_required() @require_role('admin') def upload_blueprint(levelid): """Upload this level's blueprint for one theme. multipart/form-data: file=, theme=light|dark. The native pixel size is NOT inferred from the image - it is stated on the level, because that is what existing coordinates mean and guessing it would move every marker. """ level = db.session.get(MapLevel, levelid) if not level: return error_response(ErrorCodes.NOT_FOUND, 'No such level', http_code=404) theme = (request.form.get('theme') or '').strip().lower() if theme not in ('light', 'dark'): return error_response(ErrorCodes.VALIDATION_ERROR, 'theme must be light or dark') upload = request.files.get('file') if not upload or not upload.filename: return error_response(ErrorCodes.VALIDATION_ERROR, 'No file provided') ext = os.path.splitext(upload.filename)[1].lower() if ext not in BLUEPRINT_EXTENSIONS: return error_response(ErrorCodes.VALIDATION_ERROR, f'Unsupported image type {ext}') os.makedirs(_blueprint_dir(), exist_ok=True) filename = secure_filename(f'level-{levelid}-{theme}{ext}') raw = upload.read() upload.seek(0) upload.save(os.path.join(_blueprint_dir(), filename)) url = f'/api/maplevels/{levelid}/blueprint/{filename}' setattr(level, f'blueprint{theme}', url) # The image's real pixel size, read from its header. What happens next # depends entirely on whether anything is already placed on this level. detectedwidth, detectedheight = image_size(raw) placed = Asset.query.filter_by(levelid=levelid, isactive=True).count() sizenote = None if detectedwidth and detectedheight: matches = (detectedwidth == level.mapwidth and detectedheight == level.mapheight) if matches: sizenote = None elif not placed: # Nothing is placed here yet, so no coordinate can be invalidated: # adopt the image's own size, which is almost certainly what the # operator wanted and saves them typing it. level.mapwidth = detectedwidth level.mapheight = detectedheight sizenote = ('dimensions set from the image: %d x %d' % (detectedwidth, detectedheight)) else: # Markers exist in the OLD coordinate space. Silently adopting the # new size would move every one of them relative to the drawing # while looking like a successful upload, so this reports and # changes nothing. Resizing is a transform, not an upload. sizenote = ( 'this image is %d x %d but the level is set to %d x %d, and %d ' 'marker(s) are placed in the current space. The dimensions were ' 'NOT changed: set them with a landmark transform ' '(POST /api/mappositions/transform) so the markers move with ' 'them.' % (detectedwidth, detectedheight, level.mapwidth, level.mapheight, placed)) db.session.commit() AuditLog.log('updated', 'MapLevel', entityid=levelid, entityname=level.levelname, changes={f'blueprint{theme}': {'new': url}}) db.session.commit() payload = {'levelid': levelid, f'blueprint{theme}': url, 'mapwidth': level.mapwidth, 'mapheight': level.mapheight, 'detectedwidth': detectedwidth, 'detectedheight': detectedheight, 'placedassets': placed} if sizenote: payload['sizenote'] = sizenote return success_response(payload, message='Blueprint uploaded') def _apply_default(level, requested): """Make this level the default, clearing the flag elsewhere. Exactly-one-default is a rule across rows, which no column constraint can express, so it is enforced here - the one place that sets the flag. """ if not requested: return MapLevel.query.filter(MapLevel.levelid != level.levelid).update( {'isdefault': False}) level.isdefault = True