from datetime import datetime from flask import Blueprint, flash, redirect, render_template, request, url_for from flask_login import login_required from app_new.extensions import db from sqlalchemy.exc import IntegrityError, OperationalError from app_new.core.models.college import Building, HousingUnit, HousingOccupancy, Room, RoomType, Site, Zone housing_bp = Blueprint("housing", __name__, template_folder="templates") def _date(value): return datetime.strptime(value, "%Y-%m-%d").date() if value else None @housing_bp.route("/") @login_required def index(): units = HousingUnit.query.join(Building).order_by(Building.name, HousingUnit.name).all() return render_template("housing/index.html", units=units) @housing_bp.route("/new", methods=["GET", "POST"]) @login_required def create(): housing_sites = Site.query.filter_by(site_type="logements", is_active=True).order_by(Site.name).all() buildings = Building.query.order_by(Building.name).all() if request.method == "POST": name = (request.form.get("name") or "").strip() building_id = request.form.get("building_id", type=int) if not name or not building_id: flash("Le nom et le bâtiment sont obligatoires.", "danger") return render_template("housing/form.html", unit=None, buildings=buildings, housing_sites=housing_sites), 400 unit = HousingUnit(name=name, building_id=building_id) _apply_form(unit) db.session.add(unit) try: _sync_occupancy(unit, None) db.session.commit() except OperationalError as exc: if getattr(exc.orig, 'args', [None])[0] != 1020: raise # MariaDB peut demander de rejouer une transaction lorsqu'un # parent et son historique sont créés ensemble. db.session.rollback() unit = HousingUnit(name=name, building_id=building_id) _apply_form(unit) db.session.add(unit) db.session.commit() _sync_occupancy(unit, None) db.session.commit() except IntegrityError: db.session.rollback() flash("Ce code de logement est déjà utilisé.", "danger") return render_template("housing/form.html", unit=None, buildings=buildings, housing_sites=housing_sites), 400 flash(f"Logement « {unit.name} » créé.", "success") return redirect(url_for("housing.detail", unit_id=unit.id)) return render_template("housing/form.html", unit=None, buildings=buildings, housing_sites=housing_sites) @housing_bp.route("/") @login_required def detail(unit_id): unit = HousingUnit.query.get_or_404(unit_id) equipments = [equipment for room in unit.rooms for equipment in room.equipments if not equipment.is_deleted] available_rooms = Room.query.filter_by(building_id=unit.building_id, housing_unit_id=None).order_by(Room.name).all() return render_template("housing/detail.html", unit=unit, equipments=equipments, available_rooms=available_rooms) @housing_bp.route("//edit", methods=["GET", "POST"]) @login_required def edit(unit_id): unit = HousingUnit.query.get_or_404(unit_id) buildings = Building.query.order_by(Building.name).all() if request.method == "POST": previous = (unit.occupant_name, unit.occupant_contact, unit.occupancy_start, unit.occupancy_end, unit.occupancy_status) new_building_id = request.form.get("building_id", type=int) if unit.rooms.count() and new_building_id != unit.building_id: flash("Détachez ou déplacez les pièces avant de changer de bâtiment.", "danger") return render_template("housing/form.html", unit=unit, buildings=buildings, housing_sites=[]), 400 _apply_form(unit) unit.name = (request.form.get("name") or "").strip() unit.building_id = new_building_id try: _sync_occupancy(unit, previous) db.session.commit() except IntegrityError: db.session.rollback() flash("Ce code de logement est déjà utilisé.", "danger") return render_template("housing/form.html", unit=unit, buildings=buildings, housing_sites=[]), 400 flash("Logement mis à jour.", "success") return redirect(url_for("housing.detail", unit_id=unit.id)) return render_template("housing/form.html", unit=unit, buildings=buildings, housing_sites=[]) @housing_bp.route("//rooms", methods=["POST"]) @login_required def add_room(unit_id): unit = HousingUnit.query.get_or_404(unit_id) name = (request.form.get("name") or "").strip() if not name: flash("Le nom de la pièce est obligatoire.", "danger") return redirect(url_for("housing.detail", unit_id=unit.id)) zone = Zone.query.filter_by(building_id=unit.building_id, name="Logements").first() if zone is None: zone = Zone(name="Logements", building_id=unit.building_id) db.session.add(zone) db.session.flush() room_type = RoomType.query.filter_by(name=request.form.get("room_type_name", "Logement")).first() room = Room( name=name, code=(request.form.get("code") or "").strip() or None, building_id=unit.building_id, zone_id=zone.id, room_type_id=room_type.id if room_type else None, housing_unit_id=unit.id, ) db.session.add(room) db.session.commit() flash(f"Pièce « {room.name} » ajoutée.", "success") return redirect(url_for("housing.detail", unit_id=unit.id)) @housing_bp.route("//rooms/link", methods=["POST"]) @login_required def link_rooms(unit_id): unit = HousingUnit.query.get_or_404(unit_id) room_ids = request.form.getlist("room_ids", type=int) rooms = Room.query.filter(Room.id.in_(room_ids), Room.building_id == unit.building_id, Room.housing_unit_id.is_(None)).all() if room_ids else [] for room in rooms: room.housing_unit_id = unit.id db.session.commit() flash(f"{len(rooms)} pièce(s) existante(s) rattachée(s).", "success") return redirect(url_for("housing.detail", unit_id=unit.id)) @housing_bp.route("//rooms//unlink", methods=["POST"]) @login_required def unlink_room(unit_id, room_id): room = Room.query.filter_by(id=room_id, housing_unit_id=unit_id).first_or_404() room.housing_unit_id = None db.session.commit() flash("Pièce détachée du logement sans être supprimée.", "success") return redirect(url_for("housing.detail", unit_id=unit_id)) def _apply_form(unit): unit.code = (request.form.get("code") or "").strip() or None unit.housing_type = request.form.get("housing_type", "fonction") unit.occupancy_status = request.form.get("occupancy_status", "vacant") unit.occupant_name = (request.form.get("occupant_name") or "").strip() or None unit.occupant_contact = (request.form.get("occupant_contact") or "").strip() or None unit.occupancy_start = _date(request.form.get("occupancy_start")) unit.occupancy_end = _date(request.form.get("occupancy_end")) unit.notes = (request.form.get("notes") or "").strip() or None def _sync_occupancy(unit, previous): """Conserve chaque changement d'occupant au lieu d'écraser l'historique.""" current = next((entry for entry in unit.occupancies if entry.end_date is None), None) old_name = previous[0] if previous else None changed = previous is None or old_name != unit.occupant_name or previous[2] != unit.occupancy_start if current and (changed or unit.occupancy_status != 'occupe'): current.end_date = unit.occupancy_end or datetime.now().date() current.exit_condition = (request.form.get("exit_condition") or "").strip() or None if unit.occupancy_status == 'occupe' and unit.occupant_name and (changed or current is None): unit.occupancies.append(HousingOccupancy( occupant_name=unit.occupant_name, occupant_contact=unit.occupant_contact, start_date=unit.occupancy_start or datetime.now().date(), end_date=unit.occupancy_end, key_count=max(request.form.get("key_count", type=int, default=0), 0), entry_condition=(request.form.get("entry_condition") or "").strip() or None, notes=(request.form.get("occupancy_notes") or "").strip() or None, ))