feat: introduce document sequencing service to categorize intervention files into temporal blocks and add unit tests

This commit is contained in:
kdeterme 2026-08-31 13:12:42 +02:00
parent 5e2fb023c9
commit 5f8a771277
6 changed files with 498 additions and 40 deletions

View file

@ -0,0 +1,198 @@
from typing import List, Dict, Any, Optional
from datetime import datetime
from django.utils.translation import gettext as _
from django.utils import timezone
from django.utils.dateformat import format as dj_format_date
def get_intervention_milestones(intervention) -> tuple[Optional[datetime], Optional[datetime]]:
"""
Détermine les jalons temporels d'une intervention :
- start_dt (T_début) : Début réel de l'exécution / réalisation
- end_dt (T_fin) : Fin réelle de l'exécution / réalisation
Logique de résolution :
1. start_dt :
- intervention.begin_time si défini
- Sinon premier événement timeline avec to_status='in_progress'
- Sinon premier événement InterventionRealizationTimeLine avec event_type='real_begin'
2. end_dt :
- intervention.end_time si défini
- Sinon premier événement timeline avec to_status in ['finished', 'processed', 'validated', 'invoiced', 'closed']
(intervenu après start_dt si start_dt existe)
- Sinon premier événement InterventionRealizationTimeLine avec event_type='real_end'
"""
start_dt: Optional[datetime] = intervention.begin_time or None
if not start_dt:
# Recherche dans InterventionTimeLine
first_in_progress = (
intervention.events
.filter(to_status='in_progress')
.order_by('event_time')
.first()
)
if first_in_progress:
start_dt = first_in_progress.event_time
else:
# Recherche dans InterventionRealizationTimeLine
first_real_begin = (
intervention.realized_events
.filter(event_type='real_begin')
.order_by('event_time')
.first()
)
if first_real_begin:
start_dt = first_real_begin.event_time
end_dt: Optional[datetime] = intervention.end_time or None
if not end_dt:
# Recherche dans InterventionTimeLine
terminal_statuses = ['finished', 'processed', 'validated', 'invoiced', 'closed']
events_qs = intervention.events.filter(to_status__in=terminal_statuses)
if start_dt:
events_qs = events_qs.filter(event_time__gte=start_dt)
first_terminal = events_qs.order_by('event_time').first()
if first_terminal:
end_dt = first_terminal.event_time
else:
# Recherche dans InterventionRealizationTimeLine
first_real_end = (
intervention.realized_events
.filter(event_type='real_end')
.order_by('event_time')
.first()
)
if first_real_end:
end_dt = first_real_end.event_time
return start_dt, end_dt
def get_sequenced_intervention_documents(intervention) -> Dict[str, Any]:
"""
Organise les documents et photos de l'intervention en 3 blocs temporels :
1. before_work : Avant réalisation (préparation, constat, signalement)
2. during_work : Exécution du chantier (pendant les travaux)
3. after_work : Après réalisation (contrôle, levée de réserves, clôture)
Retourne un dictionnaire :
{
'blocks': [
{
'key': 'before_work',
'title': '...',
'subtitle': '...',
'icon': '...',
'badge_class': '...',
'period_display': '...',
'modal_id': '...',
'documents': [...],
'count': int,
},
...
],
'total_count': int,
'has_documents': bool,
'start_dt': datetime|None,
'end_dt': datetime|None,
}
"""
start_dt, end_dt = get_intervention_milestones(intervention)
docs = list(
intervention.documents
.all()
.select_related('uploaded_by')
.order_by('upload_date', 'id')
)
before_docs: List[Any] = []
during_docs: List[Any] = []
after_docs: List[Any] = []
for doc in docs:
doc_dt = doc.upload_date or getattr(intervention, 'creation_time', None)
if not start_dt:
# L'intervention n'a pas encore démarré : tout est classé en avant-réalisation
before_docs.append(doc)
elif not doc_dt:
before_docs.append(doc)
elif doc_dt < start_dt:
before_docs.append(doc)
else:
# doc_dt >= start_dt
if not end_dt:
during_docs.append(doc)
elif doc_dt <= end_dt:
during_docs.append(doc)
else:
after_docs.append(doc)
# Libellés temporels formatés
def _fmt(dt: datetime) -> str:
loc_dt = timezone.localtime(dt) if timezone.is_aware(dt) else dt
return dj_format_date(loc_dt, 'd/m/Y H:i')
if start_dt:
period_before = _("Jusqu'au %(date)s") % {'date': _fmt(start_dt)}
else:
period_before = _("Phase de préparation")
if start_dt and end_dt:
period_during = _("Du %(start)s au %(end)s") % {'start': _fmt(start_dt), 'end': _fmt(end_dt)}
elif start_dt:
period_during = _("Depuis le %(date)s") % {'date': _fmt(start_dt)}
else:
period_during = ""
if end_dt:
period_after = _("Après le %(date)s") % {'date': _fmt(end_dt)}
else:
period_after = ""
blocks = [
{
'key': 'before_work',
'title': _("Avant réalisation"),
'subtitle': _("Préparation, constat initial et pièces administratives"),
'icon': 'bi-clipboard-check',
'badge_class': 'bg-secondary text-white',
'period_display': period_before,
'modal_id': 'interventionDocViewerModal_before',
'documents': before_docs,
'count': len(before_docs),
},
{
'key': 'during_work',
'title': _("Exécution du chantier"),
'subtitle': _("Travaux en cours et constat d'achèvement"),
'icon': 'bi-tools',
'badge_class': 'bg-primary text-white',
'period_display': period_during,
'modal_id': 'interventionDocViewerModal_during',
'documents': during_docs,
'count': len(during_docs),
},
{
'key': 'after_work',
'title': _("Après réalisation"),
'subtitle': _("Contrôle, levée de réserves et clôture"),
'icon': 'bi-shield-check',
'badge_class': 'bg-success text-white',
'period_display': period_after,
'modal_id': 'interventionDocViewerModal_after',
'documents': after_docs,
'count': len(after_docs),
},
]
total_count = len(docs)
return {
'blocks': blocks,
'total_count': total_count,
'has_documents': total_count > 0,
'start_dt': start_dt,
'end_dt': end_dt,
}

View file

@ -1,19 +1,47 @@
{% load document_viewer_tags i18n %}
<div id="section-documents" class="card mb-4">
<div class="card-header d-flex justify-content-between align-items-center">
<h4 class="mb-0">{% translate "Documents" %}{% if photos_required %}<span class="badge bg-warning text-dark ms-2" style="font-size:0.75rem;"><i class="bi bi-camera-fill me-1"></i>{% translate "Photos obligatoires" %}</span>{% endif %}</h4>
<h4 class="mb-0">
<i class="bi bi-images me-2 text-primary"></i>{% translate "Documents & Photos" %}{% if total_documents_count %}<span class="badge bg-secondary ms-2" style="font-size:0.75rem;">{{ total_documents_count }}</span>{% endif %}{% if photos_required %}<span class="badge bg-warning text-dark ms-2" style="font-size:0.75rem;"><i class="bi bi-camera-fill me-1"></i>{% translate "Photos obligatoires" %}</span>{% endif %}
</h4>
{% if 'documents' in allowed_update_fields %}
<button onclick="" class="btn btn-light border-0" title="Ajouter" data-bs-toggle="modal" data-bs-target="#addDocumentModal">
<button onclick="" class="btn btn-light border-0" title="{% translate 'Ajouter' %}" data-bs-toggle="modal" data-bs-target="#addDocumentModal">
<i class="bi bi-plus-circle text-primary"></i>
</button>
{% endif %}
</div>
<div class="card-body">
<div class="row">
<div id="documentsAlertContainer" class="m-1"></div>
<div id="documentsContainer">
{% render_document_viewer intervention.documents.all modal_id="interventionDocViewerModal" allow_delete=can_delete_documents delete_always_visible=can_delete_documents allow_copy=can_copy_documents %}
</div>
<div id="documentsAlertContainer" class="m-1"></div>
<div id="documentsContainer">
{% if total_documents_count == 0 %}
<div class="text-center text-muted py-4">
<i class="bi bi-images fs-2 d-block mb-2 text-secondary opacity-50"></i>
<p class="mb-0">{% translate "Aucun document ou photo pour cette intervention." %}</p>
</div>
{% else %}
{% for block in document_blocks %}
{% if block.count > 0 %}
<div class="document-block mb-4 p-3 rounded bg-light border">
<div class="d-flex justify-content-between align-items-center mb-3">
<div class="d-flex align-items-center gap-2">
<span class="badge {{ block.badge_class }} py-1 px-2" style="font-size: 0.85rem;">
<i class="bi {{ block.icon }} me-1"></i>{{ block.title }}
</span>
<span class="text-secondary small fw-medium">
{{ block.count }} {% if block.count > 1 %}{% translate "fichiers" %}{% else %}{% translate "fichier" %}{% endif %}
</span>
</div>
{% if block.period_display %}
<span class="text-muted small">
<i class="bi bi-clock me-1"></i>{{ block.period_display }}
</span>
{% endif %}
</div>
{% render_document_viewer block.documents modal_id=block.modal_id allow_delete=can_delete_documents delete_always_visible=can_delete_documents allow_copy=can_copy_documents %}
</div>
{% endif %}
{% endfor %}
{% endif %}
</div>
</div>
</div>

View file

@ -609,47 +609,78 @@
</div>
</div>
<!-- Documents -->
<!-- Documents & Photos -->
<div class="card mb-4 bg-white">
<div class="card-header d-flex justify-content-between align-items-center">
<span class="fw-semibold">{% translate "Documents" %}</span>
<span class="fw-semibold">
<i class="bi bi-images me-1 text-primary"></i>{% translate "Documents & Photos" %}
</span>
{% if total_documents_count %}
<span class="badge bg-secondary rounded-pill">{{ total_documents_count }}</span>
{% endif %}
</div>
<div class="card-body">
{% if 'documents' in allowed_update_fields %}
<div id="documentsAlertContainer" class="mb-2"></div>
{% endif %}
<div class="row" id="documentsContainer">
{% for document in intervention.documents.all %}
<div id="document-card-{{ document.id }}" class="col-6 col-md-2 mb-3" data-document-id="{{ document.id }}" data-upload-date="{{ document.upload_date|date:'c' }}">
<div class="position-relative">
{% if 'documents' in allowed_update_fields and document.id in deletable_document_ids %}
<button type="button"
class="btn btn-link text-danger p-0 position-absolute top-0 end-0 delete-document-btn bg-white"
data-document-id="{{ document.id }}"
data-document-name="{{ document.filename|default:''|escape }}"
aria-label="{% translate "Supprimer le document" %}"
title="{% translate "Supprimer le document" %}">
<i class="bi bi-trash fs-6"></i>
</button>
{% endif %}
{% private_file_url document 'file' as document_file_url %}
<a href="{{ document_file_url }}" target="_blank">
<div class="img-thumbnail" style="width: 100%; height: 100px; overflow: hidden;">
{% if document.thumbnail %}
{% private_file_url document 'thumbnail' as document_thumbnail_url %}
<img src="{{ document_thumbnail_url }}" alt="{{ document.filename }}" style="width: 100%; height: 100%; object-fit: cover; object-position: center;">
{% else %}
<img src="{{ document_file_url }}" alt="{{ document.filename }}" style="width: 100%; height: 100%; object-fit: cover; object-position: center;">
<div id="documentsContainer">
{% if total_documents_count == 0 %}
<div class="text-center text-muted py-3">
<i class="bi bi-images fs-3 d-block mb-1 text-secondary opacity-50"></i>
<span class="small">{% translate "Aucun document ou photo pour cette intervention." %}</span>
</div>
{% else %}
{% for block in document_blocks %}
{% if block.count > 0 %}
<div class="document-block-mobile mb-3 {% if not forloop.last %}border-bottom pb-3{% endif %}">
<div class="d-flex align-items-center justify-content-between mb-2">
<div class="d-flex align-items-center">
<span class="badge {{ block.badge_class }} me-2" style="font-size: 0.75rem;">
<i class="bi {{ block.icon }} me-1"></i>{{ block.title }}
</span>
<span class="text-muted small">({{ block.count }})</span>
</div>
{% if block.period_display %}
<span class="text-muted" style="font-size: 0.72rem;">{{ block.period_display }}</span>
{% endif %}
</div>
</a>
</div>
<div class="text-center text-muted mt-1" style="font-size: 0.85em;">
{{ document.upload_date|date:"d/m/Y H:i" }}<br>
{{ document.uploaded_by|short_name }}
</div>
</div>
{% endfor %}
<div class="row g-2">
{% for document in block.documents %}
<div id="document-card-{{ document.id }}" class="col-6 col-md-2 mb-2" data-document-id="{{ document.id }}" data-upload-date="{{ document.upload_date|date:'c' }}">
<div class="position-relative">
{% if 'documents' in allowed_update_fields and document.id in deletable_document_ids %}
<button type="button"
class="btn btn-link text-danger p-0 position-absolute top-0 end-0 delete-document-btn bg-white"
data-document-id="{{ document.id }}"
data-document-name="{{ document.filename|default:''|escape }}"
aria-label="{% translate "Supprimer le document" %}"
title="{% translate "Supprimer le document" %}">
<i class="bi bi-trash fs-6"></i>
</button>
{% endif %}
{% private_file_url document 'file' as document_file_url %}
<a href="{{ document_file_url }}" target="_blank">
<div class="img-thumbnail" style="width: 100%; height: 100px; overflow: hidden;">
{% if document.thumbnail %}
{% private_file_url document 'thumbnail' as document_thumbnail_url %}
<img src="{{ document_thumbnail_url }}" alt="{{ document.filename }}" style="width: 100%; height: 100%; object-fit: cover; object-position: center;">
{% else %}
<img src="{{ document_file_url }}" alt="{{ document.filename }}" style="width: 100%; height: 100%; object-fit: cover; object-position: center;">
{% endif %}
</div>
</a>
</div>
<div class="text-center text-muted mt-1" style="font-size: 0.8em;">
{{ document.upload_date|date:"d/m/Y H:i" }}<br>
{{ document.uploaded_by|short_name }}
</div>
</div>
{% endfor %}
</div>
</div>
{% endif %}
{% endfor %}
{% endif %}
</div>
</div>
</div>

View file

@ -0,0 +1,195 @@
from datetime import timedelta
from django.test import TestCase
from django.contrib.auth import get_user_model
from django.core.files.uploadedfile import SimpleUploadedFile
from django.utils import timezone
from django.urls import reverse
from common.models import UserConfig, Role, Thematic
from interventions.models import (
Intervention,
InterventionDocument,
InterventionTimeLine,
InterventionRealizationTimeLine,
)
from interventions.services.document_sequencing import (
get_intervention_milestones,
get_sequenced_intervention_documents,
)
class DocumentSequencingServiceTests(TestCase):
def setUp(self):
self.User = get_user_model()
self.user = self.User.objects.create_user(username='test_user', password='password')
self.thematic = Thematic.objects.create(code='road', name_fr='Route', name_nl='Weg')
self.now = timezone.now()
self.intervention = Intervention.objects.create(
title='Test Intervention Séquençage',
status='in_preparation',
thematic=self.thematic,
)
def test_milestones_explicit_dates(self):
t_begin = self.now
t_end = self.now + timedelta(hours=3)
self.intervention.begin_time = t_begin
self.intervention.end_time = t_end
self.intervention.save()
start_dt, end_dt = get_intervention_milestones(self.intervention)
self.assertEqual(start_dt, t_begin)
self.assertEqual(end_dt, t_end)
def test_milestones_from_timeline_events(self):
t_begin = self.now + timedelta(hours=1)
t_end = self.now + timedelta(hours=4)
InterventionTimeLine.objects.create(
intervention=self.intervention,
event_user=self.user,
event_time=t_begin,
event_type='status_change',
from_status='assigned',
to_status='in_progress',
)
InterventionTimeLine.objects.create(
intervention=self.intervention,
event_user=self.user,
event_time=t_end,
event_type='status_change',
from_status='in_progress',
to_status='finished',
)
start_dt, end_dt = get_intervention_milestones(self.intervention)
self.assertEqual(start_dt, t_begin)
self.assertEqual(end_dt, t_end)
def test_sequenced_documents_three_blocks(self):
t_begin = self.now + timedelta(hours=2)
t_end = self.now + timedelta(hours=5)
self.intervention.begin_time = t_begin
self.intervention.end_time = t_end
self.intervention.save()
dummy_file = SimpleUploadedFile("photo.jpg", b"fake image content", content_type="image/jpeg")
# Doc 1: Avant réalisation (T - 1h)
doc_before = InterventionDocument.objects.create(
intervention=self.intervention,
file=dummy_file,
uploaded_by=self.user,
)
InterventionDocument.objects.filter(id=doc_before.id).update(upload_date=self.now + timedelta(hours=1))
# Doc 2: Pendant exécution (T + 3h)
doc_during = InterventionDocument.objects.create(
intervention=self.intervention,
file=dummy_file,
uploaded_by=self.user,
)
InterventionDocument.objects.filter(id=doc_during.id).update(upload_date=self.now + timedelta(hours=3))
# Doc 3: Après réalisation (T + 6h)
doc_after = InterventionDocument.objects.create(
intervention=self.intervention,
file=dummy_file,
uploaded_by=self.user,
)
InterventionDocument.objects.filter(id=doc_after.id).update(upload_date=self.now + timedelta(hours=6))
result = get_sequenced_intervention_documents(self.intervention)
self.assertEqual(result['total_count'], 3)
self.assertTrue(result['has_documents'])
blocks_map = {b['key']: b for b in result['blocks']}
self.assertEqual(blocks_map['before_work']['count'], 1)
self.assertEqual(blocks_map['before_work']['documents'][0].id, doc_before.id)
self.assertEqual(blocks_map['during_work']['count'], 1)
self.assertEqual(blocks_map['during_work']['documents'][0].id, doc_during.id)
self.assertEqual(blocks_map['after_work']['count'], 1)
self.assertEqual(blocks_map['after_work']['documents'][0].id, doc_after.id)
def test_sequenced_documents_before_start_only(self):
# Intervention pas encore démarrée
self.intervention.begin_time = None
self.intervention.end_time = None
self.intervention.save()
dummy_file = SimpleUploadedFile("doc.pdf", b"pdf content", content_type="application/pdf")
doc = InterventionDocument.objects.create(
intervention=self.intervention,
file=dummy_file,
uploaded_by=self.user,
)
result = get_sequenced_intervention_documents(self.intervention)
self.assertEqual(result['total_count'], 1)
blocks_map = {b['key']: b for b in result['blocks']}
self.assertEqual(blocks_map['before_work']['count'], 1)
self.assertEqual(blocks_map['during_work']['count'], 0)
self.assertEqual(blocks_map['after_work']['count'], 0)
class DocumentSequencingViewsTests(TestCase):
def setUp(self):
self.User = get_user_model()
self.user = self.User.objects.create_user(username='admin_user', password='password')
self.role_admin = Role.objects.create(name='admin')
self.user_config = UserConfig.objects.create(user=self.user, is_intern=True)
self.user_config.roles.add(self.role_admin)
self.thematic = Thematic.objects.create(code='road', name_fr='Route', name_nl='Weg')
self.now = timezone.now()
self.intervention = Intervention.objects.create(
title='Intervention vue séquencée',
status='in_progress',
thematic=self.thematic,
begin_time=self.now - timedelta(hours=2),
)
dummy_file = SimpleUploadedFile("photo1.jpg", b"fake photo", content_type="image/jpeg")
doc1 = InterventionDocument.objects.create(
intervention=self.intervention,
file=dummy_file,
uploaded_by=self.user,
)
InterventionDocument.objects.filter(id=doc1.id).update(upload_date=self.now - timedelta(hours=3))
doc2 = InterventionDocument.objects.create(
intervention=self.intervention,
file=dummy_file,
uploaded_by=self.user,
)
InterventionDocument.objects.filter(id=doc2.id).update(upload_date=self.now - timedelta(hours=1))
self.client.force_login(self.user)
def test_desktop_detail_view_contains_sequenced_blocks(self):
url = reverse('interventions:intervention_detail', args=[self.intervention.id])
response = self.client.get(url + '?force_desktop=1')
self.assertEqual(response.status_code, 200)
self.assertIn('document_blocks', response.context)
self.assertEqual(response.context['total_documents_count'], 2)
content = response.content.decode('utf-8')
self.assertIn('Avant réalisation', content)
self.assertIn('Exécution du chantier', content)
def test_mobile_detail_view_contains_sequenced_blocks(self):
url = reverse('interventions:intervention_detail_mobile', args=[self.intervention.id])
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertIn('document_blocks', response.context)
self.assertEqual(response.context['total_documents_count'], 2)
content = response.content.decode('utf-8')
self.assertIn('Avant réalisation', content)
self.assertIn('Exécution du chantier', content)

View file

@ -62,6 +62,7 @@ from interventions.models import (InterventionSubscription, Intervention, Interv
GuidedOperationTemplate, GuidedOperationData, GuidedOperationMedia,
InterventionPostPeriodQuantity, InterventionEmailLog, InterventionAssignment)
from interventions.models import STATUS_CHOICES, STATUS_ORDERS, MAINTAIN_CHOICES, TYPE_CHOICES, ORIGIN_TYPE_CHOICES, PAUSE_REASON_CHOICES, get_type_choices
from interventions.services.document_sequencing import get_sequenced_intervention_documents
from interventions.views.forms import (InterventionForm, InterventionNoteForm, InterventionDocumentForm,
InterventionOccupationForm, PreventiveInterventionForm, AssetsLinkForm,
LinkExistingInterventionsForm)
@ -741,7 +742,9 @@ def intervention_detail(request, intervention_id):
# Discussion
'discussion_messages': discussion_messages,
'can_post_discussion_message': can_post_discussion_message,
# Document permissions
# Document permissions & sequenced blocks
'document_blocks': get_sequenced_intervention_documents(intervention)['blocks'],
'total_documents_count': intervention.documents.count(),
'can_delete_documents': 'documents' in allowed_update_fields,
'can_copy_documents': 'documents' in allowed_update_fields and copy_targets_json != '[]',
# Guided operation template

View file

@ -64,6 +64,7 @@ from interventions.models import (InterventionSubscription, Intervention, Interv
InterventionPostPeriodQuantity, InterventionEmailLog, InterventionAssignment)
from interventions import views
from interventions.models import STATUS_CHOICES, STATUS_ORDERS, MAINTAIN_CHOICES, TYPE_CHOICES, ORIGIN_TYPE_CHOICES, PAUSE_REASON_CHOICES
from interventions.services.document_sequencing import get_sequenced_intervention_documents
from interventions.views.forms import (InterventionForm, InterventionNoteForm, InterventionDocumentForm,
InterventionOccupationForm, PreventiveInterventionForm, AssetsLinkForm,
LinkExistingInterventionsForm)
@ -1074,6 +1075,8 @@ def intervention_detail_mobile(request, intervention_id):
context = {
'intervention': intervention,
'document_blocks': get_sequenced_intervention_documents(intervention)['blocks'],
'total_documents_count': intervention.documents.count(),
'assets': assets,
'notes': notes,
'can_edit_intervention': can_edit_mobile,