283 lines
13 KiB
Python
283 lines
13 KiB
Python
"""
|
||
Contracts Routes - GMAO College
|
||
Suivi des contrats d'entreprise.
|
||
"""
|
||
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify
|
||
from flask_login import login_required, current_user
|
||
from ..extensions import db
|
||
from ..contracts.models import Contract, ContractVisit
|
||
from ..core.models.maintenance import Lot
|
||
from ..core.models.equipment import Equipment
|
||
from ..companies.routes import Company
|
||
from datetime import datetime, date, timedelta, timezone, time
|
||
|
||
def _optional_date(value):
|
||
return datetime.strptime(value, '%Y-%m-%d').date() if value else None
|
||
|
||
|
||
def _optional_time(value):
|
||
return datetime.strptime(value, '%H:%M').time() if value else None
|
||
|
||
def _apply_visits(contract):
|
||
contract.visit_interval_days = request.form.get('visit_interval_days', type=int)
|
||
contract.last_visit_date = _optional_date(request.form.get('last_visit_date'))
|
||
contract.next_visit_date = _optional_date(request.form.get('next_visit_date'))
|
||
|
||
|
||
def _generate_visits(contract, horizon_days=365):
|
||
"""Matérialise le calendrier contractuel sans créer de doublons."""
|
||
if not contract.visit_interval_days or contract.visit_interval_days <= 0:
|
||
return 0
|
||
cursor = contract.next_visit_date or contract.start_date
|
||
if contract.last_visit_date and cursor <= contract.last_visit_date:
|
||
cursor = contract.last_visit_date + timedelta(days=contract.visit_interval_days)
|
||
horizon = min(date.today() + timedelta(days=horizon_days), contract.end_date) if contract.end_date else date.today() + timedelta(days=horizon_days)
|
||
existing = {visit.scheduled_date for visit in contract.visits}
|
||
created = 0
|
||
while cursor and cursor <= horizon:
|
||
if cursor >= contract.start_date and cursor not in existing:
|
||
db.session.add(ContractVisit(contract=contract, scheduled_date=cursor))
|
||
created += 1
|
||
cursor += timedelta(days=contract.visit_interval_days)
|
||
return created
|
||
|
||
contracts_bp = Blueprint('contracts', __name__, url_prefix='/contracts', template_folder='templates')
|
||
|
||
|
||
@contracts_bp.route('/')
|
||
@login_required
|
||
def index():
|
||
"""Liste de tous les contrats."""
|
||
contracts = Contract.query.order_by(
|
||
db.case((Contract.end_date.is_(None), 1), else_=0),
|
||
Contract.end_date.asc()
|
||
).all()
|
||
|
||
# Stats
|
||
active = sum(1 for c in contracts if c.is_active)
|
||
expiring = sum(1 for c in contracts if c.is_expiring_soon)
|
||
notice = sum(1 for c in contracts if c.must_send_notice_now)
|
||
expired = sum(1 for c in contracts if c.status == 'expire' or (c.end_date and c.end_date < date.today() and c.status == 'actif'))
|
||
|
||
return render_template('contracts/index.html',
|
||
contracts=contracts,
|
||
active_count=active,
|
||
expiring_count=expiring,
|
||
notice_count=notice,
|
||
expired_count=expired)
|
||
|
||
|
||
@contracts_bp.route('/new', methods=['GET', 'POST'])
|
||
@login_required
|
||
def create():
|
||
"""Creer un contrat."""
|
||
lots = Lot.query.order_by(Lot.name).all()
|
||
equipments = Equipment.query.order_by(Equipment.name).all()
|
||
companies = Company.query.order_by(Company.name).all()
|
||
|
||
if request.method == 'POST':
|
||
contract = Contract(
|
||
name=request.form.get('name'),
|
||
contract_number=request.form.get('contract_number'),
|
||
lot_id=request.form.get('lot_id') or None,
|
||
equipment_id=request.form.get('equipment_id') or None,
|
||
company_id=request.form.get('company_id') or None,
|
||
signed_date=datetime.strptime(request.form.get('signed_date'), '%Y-%m-%d').date() if request.form.get('signed_date') else None,
|
||
start_date=datetime.strptime(request.form.get('start_date'), '%Y-%m-%d').date() if request.form.get('start_date') else date.today(),
|
||
end_date=datetime.strptime(request.form.get('end_date'), '%Y-%m-%d').date() if request.form.get('end_date') else None,
|
||
tacit_renewal=request.form.get('tacit_renewal') == '1',
|
||
renewal_period_days=int(request.form.get('renewal_period_days', 365)),
|
||
notice_period_days=int(request.form.get('notice_period_days')) if request.form.get('notice_period_days') else None,
|
||
notice_deadline=datetime.strptime(request.form.get('notice_deadline'), '%Y-%m-%d').date() if request.form.get('notice_deadline') else None,
|
||
annual_amount=float(request.form.get('annual_amount')) if request.form.get('annual_amount') else None,
|
||
status=request.form.get('status', 'actif'),
|
||
notes=request.form.get('notes', ''),
|
||
)
|
||
db.session.add(contract)
|
||
_apply_visits(contract)
|
||
db.session.flush()
|
||
_generate_visits(contract)
|
||
db.session.commit()
|
||
flash('Contrat cree avec succes.', 'success')
|
||
return redirect(url_for('contracts.detail', id=contract.id))
|
||
|
||
# Pre-remplir si lot_id ou equipment_id dans l'URL
|
||
pre_lot = request.args.get('lot_id', type=int)
|
||
pre_eq = request.args.get('equipment_id', type=int)
|
||
pre_company = request.args.get('company_id', type=int)
|
||
|
||
return render_template('contracts/form.html',
|
||
lots=lots, equipments=equipments, companies=companies,
|
||
pre_lot=pre_lot, pre_eq=pre_eq, pre_company=pre_company,
|
||
contract=None)
|
||
|
||
|
||
@contracts_bp.route('/<int:id>')
|
||
@login_required
|
||
def detail(id):
|
||
"""Detail d'un contrat."""
|
||
contract = Contract.query.get_or_404(id)
|
||
from app_new.core.models.maintenance import LotTask
|
||
from app_new.core.models.planning import ScheduledTask
|
||
covered_tasks = LotTask.query.filter_by(contract_id=contract.id).order_by(LotTask.tache).all()
|
||
covered_task_ids = [task.id for task in covered_tasks]
|
||
covered_occurrences = (ScheduledTask.query.filter(ScheduledTask.lot_task_id.in_(covered_task_ids))
|
||
.order_by(ScheduledTask.scheduled_date.desc()).all()) if covered_task_ids else []
|
||
return render_template('contracts/detail.html', contract=contract,
|
||
covered_tasks=covered_tasks, covered_occurrences=covered_occurrences)
|
||
|
||
|
||
@contracts_bp.route('/<int:id>/edit', methods=['GET', 'POST'])
|
||
@login_required
|
||
def edit(id):
|
||
"""Modifier un contrat."""
|
||
contract = Contract.query.get_or_404(id)
|
||
lots = Lot.query.order_by(Lot.name).all()
|
||
equipments = Equipment.query.order_by(Equipment.name).all()
|
||
companies = Company.query.order_by(Company.name).all()
|
||
|
||
if request.method == 'POST':
|
||
contract.name = request.form.get('name')
|
||
contract.contract_number = request.form.get('contract_number')
|
||
contract.lot_id = request.form.get('lot_id') or None
|
||
contract.equipment_id = request.form.get('equipment_id') or None
|
||
contract.company_id = request.form.get('company_id') or None
|
||
contract.signed_date = datetime.strptime(request.form.get('signed_date'), '%Y-%m-%d').date() if request.form.get('signed_date') else None
|
||
contract.start_date = datetime.strptime(request.form.get('start_date'), '%Y-%m-%d').date() if request.form.get('start_date') else date.today()
|
||
contract.end_date = datetime.strptime(request.form.get('end_date'), '%Y-%m-%d').date() if request.form.get('end_date') else None
|
||
contract.tacit_renewal = request.form.get('tacit_renewal') == '1'
|
||
contract.renewal_period_days = int(request.form.get('renewal_period_days', 365))
|
||
contract.notice_period_days = int(request.form.get('notice_period_days')) if request.form.get('notice_period_days') else None
|
||
contract.notice_deadline = datetime.strptime(request.form.get('notice_deadline'), '%Y-%m-%d').date() if request.form.get('notice_deadline') else None
|
||
contract.annual_amount = float(request.form.get('annual_amount')) if request.form.get('annual_amount') else None
|
||
contract.status = request.form.get('status', 'actif')
|
||
contract.notes = request.form.get('notes', '')
|
||
_apply_visits(contract)
|
||
_generate_visits(contract)
|
||
db.session.commit()
|
||
flash('Contrat modifie.', 'success')
|
||
return redirect(url_for('contracts.detail', id=contract.id))
|
||
|
||
return render_template('contracts/form.html',
|
||
contract=contract, lots=lots, equipments=equipments, companies=companies,
|
||
pre_lot=None, pre_eq=None, pre_company=None)
|
||
|
||
|
||
@contracts_bp.route('/<int:id>/delete', methods=['POST'])
|
||
@login_required
|
||
def delete(id):
|
||
"""Supprimer un contrat."""
|
||
contract = Contract.query.get_or_404(id)
|
||
db.session.delete(contract)
|
||
db.session.commit()
|
||
flash('Contrat supprime.', 'success')
|
||
return redirect(url_for('contracts.index'))
|
||
|
||
|
||
@contracts_bp.route('/<int:id>/visits/generate', methods=['POST'])
|
||
@login_required
|
||
def generate_visits(id):
|
||
contract = Contract.query.get_or_404(id)
|
||
count = _generate_visits(contract)
|
||
db.session.commit()
|
||
flash(f'{count} visite(s) ajoutée(s) au calendrier.', 'success')
|
||
return redirect(url_for('contracts.detail', id=id))
|
||
|
||
|
||
@contracts_bp.route('/<int:id>/visits/<int:visit_id>/complete', methods=['POST'])
|
||
@login_required
|
||
def complete_visit(id, visit_id):
|
||
contract = Contract.query.get_or_404(id)
|
||
visit = ContractVisit.query.filter_by(id=visit_id, contract_id=id).first_or_404()
|
||
visit.status = 'realisee'
|
||
visit.completed_at = datetime.now(timezone.utc)
|
||
visit.actual_date = date.today()
|
||
visit.report = (request.form.get('report') or '').strip() or None
|
||
contract.last_visit_date = visit.actual_date
|
||
contract.next_visit_date = visit.actual_date + timedelta(days=contract.visit_interval_days) if contract.visit_interval_days else None
|
||
_generate_visits(contract)
|
||
db.session.commit()
|
||
flash('Visite extérieure marquée comme réalisée.', 'success')
|
||
return redirect(url_for('contracts.detail', id=id))
|
||
|
||
|
||
@contracts_bp.route('/<int:id>/visits/new', methods=['POST'])
|
||
@login_required
|
||
def create_visit(id):
|
||
"""Enregistre un passage sans inventer d'heure pour les modes date libre."""
|
||
contract = Contract.query.get_or_404(id)
|
||
scheduled = _optional_date(request.form.get('scheduled_date'))
|
||
mode = request.form.get('passage_mode') or 'appointment'
|
||
if not scheduled:
|
||
flash('La date prévue est obligatoire.', 'danger')
|
||
return redirect(url_for('contracts.detail', id=id))
|
||
if mode not in ('appointment', 'date_only', 'open_window'):
|
||
flash('Mode de passage invalide.', 'danger')
|
||
return redirect(url_for('contracts.detail', id=id))
|
||
visit = ContractVisit(
|
||
contract=contract, scheduled_date=scheduled, passage_mode=mode,
|
||
start_time=_optional_time(request.form.get('start_time')) if mode == 'appointment' else None,
|
||
end_time=_optional_time(request.form.get('end_time')) if mode == 'appointment' else None,
|
||
window_start=_optional_date(request.form.get('window_start')) if mode == 'open_window' else None,
|
||
window_end=_optional_date(request.form.get('window_end')) if mode == 'open_window' else None,
|
||
accompaniment_mode=request.form.get('accompaniment_mode') or 'none',
|
||
accompaniment_minutes=request.form.get('accompaniment_minutes', type=int),
|
||
technician_id=request.form.get('technician_id', type=int) or None,
|
||
lot_task_id=request.form.get('lot_task_id', type=int) or None,
|
||
status='planifiee' if mode != 'open_window' else 'passage_libre',
|
||
)
|
||
if mode == 'open_window' and visit.window_start is None:
|
||
visit.window_start = scheduled
|
||
db.session.add(visit)
|
||
db.session.commit()
|
||
flash('Passage enregistré.', 'success')
|
||
return redirect(url_for('contracts.detail', id=id))
|
||
|
||
|
||
@contracts_bp.route('/<int:id>/visits/<int:visit_id>/no-show', methods=['POST'])
|
||
@login_required
|
||
def no_show_visit(id, visit_id):
|
||
visit = ContractVisit.query.filter_by(id=visit_id, contract_id=id).first_or_404()
|
||
visit.status = 'non_venue'
|
||
visit.report = (request.form.get('report') or '').strip() or visit.report
|
||
db.session.commit()
|
||
flash('Entreprise marquée non venue : la maintenance reste à réaliser.', 'warning')
|
||
return redirect(url_for('contracts.detail', id=id))
|
||
|
||
|
||
@contracts_bp.route('/<int:id>/visits/<int:visit_id>/arrived', methods=['POST'])
|
||
@login_required
|
||
def arrived_visit(id, visit_id):
|
||
visit = ContractVisit.query.filter_by(id=visit_id, contract_id=id).first_or_404()
|
||
visit.status = 'entreprise_arrivee'
|
||
visit.arrival_at = datetime.now(timezone.utc)
|
||
db.session.commit()
|
||
flash('Arrivée de l’entreprise enregistrée.', 'success')
|
||
return redirect(url_for('contracts.detail', id=id))
|
||
|
||
|
||
@contracts_bp.route('/visits')
|
||
@login_required
|
||
def visits_dashboard():
|
||
"""Vue métier des passages et contrôles externalisés."""
|
||
status = request.args.get('status')
|
||
query = ContractVisit.query.join(Contract)
|
||
if status:
|
||
query = query.filter(ContractVisit.status == status)
|
||
visits = query.order_by(ContractVisit.scheduled_date.asc()).all()
|
||
today = date.today()
|
||
return render_template('contracts/visits.html', visits=visits, today=today, status=status)
|
||
|
||
|
||
@contracts_bp.route('/api/summary')
|
||
@login_required
|
||
def api_summary():
|
||
"""Resume des contrats pour le dashboard."""
|
||
contracts = Contract.query.filter_by(status='actif').all()
|
||
return jsonify({
|
||
'total': len(contracts),
|
||
'active': sum(1 for c in contracts if c.is_active),
|
||
'expiring_soon': sum(1 for c in contracts if c.is_expiring_soon),
|
||
'must_send_notice': sum(1 for c in contracts if c.must_send_notice_now),
|
||
'expired': sum(1 for c in contracts if c.end_date and c.end_date < date.today()),
|
||
})
|