loko/loko/interventions/views/helpers.py

3803 lines
140 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from django.http import HttpResponse, JsonResponse, HttpResponseBadRequest, HttpResponseForbidden, HttpResponseRedirect
from django.shortcuts import render, get_object_or_404, redirect
from django.template.loader import render_to_string
from django.urls import reverse, NoReverseMatch
from django.core.serializers import serialize
from django.core.exceptions import FieldDoesNotExist
from django.core.files.base import ContentFile
from django.contrib.gis.db.models.functions import Transform, Centroid, AsGeoJSON, Distance
from django.contrib.gis.geos import GEOSGeometry, Point, MultiPolygon
from django.contrib.gis.measure import D
from django.contrib.contenttypes.models import ContentType
from django.contrib import messages
from django.views.decorators.http import require_POST, require_GET, require_http_methods
from django.contrib.auth.decorators import login_required
from django.utils.timezone import now, localtime, make_aware, get_current_timezone, is_naive
from django.utils import timezone as django_timezone
from django.utils import timezone
from django.utils.dateparse import parse_datetime
from django.utils.dateformat import format as dj_format_date
from django.utils.text import Truncator
from django.utils.html import strip_tags
from django.template.defaultfilters import linebreaksbr
from django.utils.translation import get_language, gettext as _, gettext_lazy as _lazy, ngettext
from django.apps import apps
from django.conf import settings
from django.core.mail import EmailMultiAlternatives
from django.db.models import F, DateTimeField, Q, Case, When, IntegerField, CharField, Value, Count, Prefetch, Sum, Max, QuerySet, OuterRef, Subquery, Exists
from django.db.models.functions import Coalesce, Lower, Cast
from django.db.models.expressions import ExpressionWrapper
from django.db.models import DurationField
from django.db.models.fields.related import ForeignKey
from django.db import transaction
from collections import defaultdict, OrderedDict
from decimal import Decimal, InvalidOperation
import json
import os
import io
import zipfile
from fpdf import FPDF
import pytz
from urllib.parse import urlencode
import re
from html import unescape
from typing import List, Iterable, Tuple, Dict, Optional, Set
import mimetypes
import csv
from io import BytesIO
from datetime import timedelta, datetime
from dateutil.relativedelta import relativedelta
from common.private_files.helpers import build_private_url
from common.utils import get_short_name
from interventions.models import (InterventionSubscription, Intervention, InterventionAsset, InterventionTimeLine, InterventionContractPost, InterventionDocument, InterventionNote, InterventionLocation,
InterventionLink,
ContractPauseReasonAssignment,
Symptom, SourceCategory, InterventionPlanificationTimeLine, InterventionRealizationTimeLine,
InterventionOccupation,
Operation, OperationTemplate, OperationTemplateItem, OperationStatus, OperationStatusSet, MeasurementDefinition,
GuidedOperationTemplate, GuidedOperationData, GuidedOperationMedia,
ContractOperationTemplatePost,
InterventionPostPeriodQuantity, InterventionEmailLog, InterventionAssignment,
DirectContractRouting)
from interventions.models import STATUS_CHOICES, STATUS_ORDERS, MAINTAIN_CHOICES, TYPE_CHOICES, ORIGIN_TYPE_CHOICES, PAUSE_REASON_CHOICES, get_type_choices
from interventions.views.forms import (InterventionForm, InterventionNoteForm, InterventionDocumentForm,
InterventionOccupationForm, PreventiveInterventionForm, AssetsLinkForm,
LinkExistingInterventionsForm)
from interventions.permissions import (intervention_permission_required, intervention_add_permission_required, intervention_edit_permission_required,
intervention_add_preventive_permission_required,
get_allowed_statuses_for_user, get_allowed_update_fields,
can_view_intervention, can_control_intervention, can_edit_intervention,
can_view_intervention_summary,
get_accessible_interventions_for_technician, filter_viewable_interventions_for_user,
can_edit_note, can_delete_document, can_manage_occupations,
can_add_intervention, can_add_repair_intervention,
intervention_manage_occupations_required,
get_contracts_for_intervention_creation, is_external_manager_with_contract_creation_rights,
should_show_created_by_provider_badge, can_view_correction_messages)
from interventions.permissions import ALLOWED_TRANSITIONS, get_allowed_transitions, user_allowed_transitions
from interventions.views.drafts import InterventionDraftManager
from interventions.utils import resolve_intervention_localisations
from common.models import UserConfig, UserContractAccess, UserContractStatusPermission, Thematic, UserThematics
from contracts.models import ContractOrder, ContractPost, Contract, Company, CompanyTeam, CompanyMember, UserCompanyTeamAccess, CompanyMemberHistory
from contracts.permissions import get_allowed_contracts_for_user, user_has_access_to_contract_order
from projects.models import Project, ProjectIntervention
from assets.models import (
PublicLightingStreet,
TrafficLightIntersection, TrafficLightContract,
NatureLocation,
StructureLocation, Structure, StructureContract,
RoadStreet,
SignStreet,
AssetCategory,
AbstractAsset,
AbstractAssetModel,
ITSGeoAsset, ITSGeoAssetContract,
ITSLocation, ITSLocationContract,
CleanLocation, CleanLocationContract,
)
from assets.permissions import get_allowed_actions_for_asset
from assets.views.locations import LOCATION_REGISTRY, registry_items_for_thematics, base_search_qs, SEARCH_LIMIT_PER_TYPE, SEARCH_LIMIT_TOTAL
from common.utils import redirect_with_variant, measure_time, strip_accents_lower
from observations.permissions import get_observation_access_context
from observations.models import Observation as ObservationModel
from controls.permissions import user_can_view_control
from interventions.views.views_pdf_reporting import (
generate_intervention_pdf,
preview_intervention_pdf,
download_intervention_report_pdf,
preview_intervention_report_pdf,
bulk_execution_report_pdf,
start_bulk_execution_report,
poll_bulk_execution_report,
download_bulk_execution_report,
)
import logging
import smtplib
import socket
import ssl
logger = logging.getLogger(__name__)
__all__ = [
'CANDIDATE_LOCATION_FKS',
'HTML_BREAK_RE',
'HTML_PARAGRAPH_CLOSE_RE',
'HTML_PARAGRAPH_OPEN_RE',
'LINKED_RECENT_STATUSES',
'LINKED_TARGET_STATUSES',
'MAX_FEATURES',
'MAX_MOBILE_ASSET_RESULTS',
'MEMBER_VIEW_ALLOWED_ROLES',
'PAUSE_REASON_MAP',
'_ASSET_CLASS_CACHE',
'_ASSET_MODEL_CLASS_CACHE',
'_add_operations_from_templates',
'_allowed_ct_ids_from_thematics',
'_apply_planification_side_effects',
'_apply_realization_side_effects',
'_asset_matches_filters',
'_asset_matches_template_filters',
'_assign_contract_posts_from_templates',
'_assign_guided_template',
'_assign_team_from_templates',
'_attach_equipment_templates',
'_build_guided_media_json',
'_build_link_groups',
'_build_replacement_model_options',
'_can_move_interventions_in_calendar',
'_can_use_member_view',
'_clear_planification_side_effects',
'_clear_realization_side_effects',
'_collect_assets_from_categories',
'_collect_assets_from_locations',
'_collect_notes_plain_text',
'_create_equipment_operations_for_assets',
'_delete_operations_for_assets',
'_expand_csv_params',
'_expand_hierarchical_location_ids',
'_extract_assets_from_location',
'_flatten_asset_collection',
'_force_delete_operations_for_assets',
'_format_location_label',
'_get_allowed_asset_category_ids',
'_get_asset_classes',
'_get_asset_model_classes',
'_get_available_occupation_members',
'_get_equipment_templates_for_intervention',
'_get_photos_required_since',
'_is_photo_required_for_user',
'_get_role_names',
'_get_templates_with_equipment_filters',
'_get_user_short_name',
'_group_operations_by_asset_and_template',
'_handle_notes_update',
'_handle_planification_update',
'_handle_realization_update',
'_html_to_plain_text',
'_log_planning_change',
'_mark_asset_as_removed',
'_member_view_allowed_company_ids',
'_member_view_allowed_contract_ids',
'_member_view_allowed_team_ids',
'_member_view_allowed_thematic_ids',
'_parse_int_query_values',
'_parse_location_ct_obj_params',
'_parse_model_filters',
'_parse_pairs_json',
'_parse_selected_locations',
'_registry_by_ct',
'_resolve_assets_from_pairs',
'_resolve_location_pairs_to_links',
'_resolve_locations_payload',
'_search_mobile_assets',
'_serialize_asset_entry',
'_serialize_note',
'_serialize_occupation',
'_serialize_timeline_event',
'_sync_intervention_assets_data',
'_unlink_operations_from_assets',
'_update_intervention_field',
'annotate_preparation_status',
'apply_intervention_filters_from_querydict',
'apply_location_filter',
'auto_check_if_eligible',
'batch_fetch_assets_with_locations',
'format_pause_description',
'get_contract_cc_recipients',
'get_contract_recipients',
'get_intervention_attachments',
'inspection_configuration_geojson',
'is_smtp_available',
'is_user_same_as_provider',
'resolve_asset_category',
'resolve_asset_spatial_data',
'resolve_coords_from_hierarchy',
'resolve_intervention_contract_and_provider',
'resolve_selected_assets',
'to_multipolygon',
'update_operation_comment'
]
PAUSE_REASON_MAP = dict(PAUSE_REASON_CHOICES)
MAX_MOBILE_ASSET_RESULTS = 100
_ASSET_CLASS_CACHE: Optional[List[type]] = None
_ASSET_MODEL_CLASS_CACHE: Optional[List[type]] = None
MEMBER_VIEW_ALLOWED_ROLES: Set[str] = {'admin', 'manager', 'controller', 'external_manager'}
HTML_BREAK_RE = re.compile(r"<\s*br\s*/?>", re.IGNORECASE)
HTML_PARAGRAPH_CLOSE_RE = re.compile(r"</\s*p\s*>", re.IGNORECASE)
HTML_PARAGRAPH_OPEN_RE = re.compile(r"<\s*p[^>]*?>", re.IGNORECASE)
MAX_FEATURES = 5000
CANDIDATE_LOCATION_FKS = ("intersection", "street", "section", "location", "structure")
LINKED_TARGET_STATUSES = ['to_be_approved', 'to_be_planned', 'to_be_processed', 'assigned', 'in_progress', 'on_pause']
LINKED_RECENT_STATUSES = ['finished', 'processed', 'corrected', 'to_be_corrected', 'validated', 'invoiced', 'closed', 'canceled']
logger = logging.getLogger(__name__)
def _get_role_names(user_config: Optional[UserConfig]) -> Set[str]:
if not user_config:
return set()
return {role.name for role in user_config.roles.all()}
def _can_use_member_view(role_names: Set[str]) -> bool:
return bool(role_names & MEMBER_VIEW_ALLOWED_ROLES)
def _can_move_interventions_in_calendar(role_names: Set[str], view_mode: str) -> bool:
"""Détermine si l'utilisateur peut déplacer des interventions via drag & drop.
- Dans la vue membres et équipes quotidien: seuls admin, top_manager, controller, external_manager peuvent déplacer (pas les managers)
- Dans la vue timeline: admin, top_manager, manager, controller, external_manager peuvent déplacer
(mais les managers ont des restrictions par statut, gérées ailleurs)
"""
if view_mode in ('members', 'teams_daily'):
# Vue membres & quotidien: managers ne peuvent PAS déplacer
return bool(role_names & {'admin', 'top_manager', 'controller', 'external_manager'})
else:
# Vue timeline: managers peuvent déplacer (avec restrictions par statut)
return bool(role_names & {'admin', 'top_manager', 'manager', 'controller', 'external_manager'})
def _member_view_allowed_company_ids(
user_config: Optional[UserConfig],
role_names: Set[str],
) -> Optional[Set[int]]:
"""Return company IDs whose teams can appear in member view."""
if not user_config:
return set()
if 'admin' in role_names:
return None # Admins can see every team.
# En vue membres, la visibilité des équipes repose sur un droit explicite
# de consultation des interventions sur le contrat.
company_ids = UserContractAccess.objects.filter(
user_config=user_config,
can_view_interventions=True,
).values_list('contract__company_id', flat=True)
return {company_id for company_id in company_ids if company_id is not None}
def _member_view_allowed_thematic_ids(
user_config: Optional[UserConfig],
role_names: Set[str],
) -> Optional[Set[int]]:
"""Return thematic IDs for which the user can view interventions.
Teams will be filtered to only show those associated with these thematics.
"""
if not user_config:
return set()
if 'admin' in role_names:
return None # Admins can see every team.
thematic_ids = UserThematics.objects.filter(
user_config=user_config,
can_view_interventions=True,
).values_list('thematic_id', flat=True)
return set(thematic_ids)
def _member_view_allowed_contract_ids(
user_config: Optional[UserConfig],
role_names: Set[str],
) -> Optional[Set[int]]:
"""Return contract IDs for which the user can change status to 'to_be_processed' or 'assigned',
or contracts accessible to external managers.
Used to restrict the backlog (right panel) to interventions the user could
actually assign or plan.
"""
if not user_config:
return set()
if 'admin' in role_names:
return None # Admins see all contracts.
contract_ids = UserContractAccess.objects.filter(
user_config=user_config,
can_view_interventions=True,
).filter(
Q(status_permissions__status__in=['to_be_processed', 'assigned', 'to_be_planned', 'in_progress'],
status_permissions__can_change_status_to=True)
| Q(user_config__roles__name='external_manager')
).values_list('contract_id', flat=True)
return {cid for cid in contract_ids if cid is not None}
def _member_view_allowed_team_ids(
user_config: Optional[UserConfig],
role_names: Set[str],
) -> Optional[Set[int]]:
"""Return explicit team IDs visible in member view.
Visibility is driven by UserCompanyTeamAccess.can_view.
"""
if not user_config:
return set()
if 'admin' in role_names:
return None # Admins can see every team.
team_ids = UserCompanyTeamAccess.objects.filter(
user_config=user_config,
can_view=True,
).values_list('team_id', flat=True)
return set(team_ids)
def _html_to_plain_text(value: str) -> str:
"""Convert limited HTML content to a readable plain text representation."""
if not value:
return ""
# First decode HTML entities before any other processing
text = unescape(value)
text = HTML_BREAK_RE.sub("\n", text)
text = HTML_PARAGRAPH_CLOSE_RE.sub("\n\n", text)
text = HTML_PARAGRAPH_OPEN_RE.sub("", text)
text = strip_tags(text)
# Decode again in case strip_tags re-encoded something
text = unescape(text)
text = text.replace("\r\n", "\n").replace("\r", "\n")
text = re.sub(r"\n{3,}", "\n\n", text)
return text.strip()
def _collect_notes_plain_text(intervention: Intervention) -> list[str]:
"""Return decoded note contents ready for insertion into plain text emails."""
try:
notes_qs = intervention.notes.all().order_by('note_time')
except Exception:
return []
collected = []
for note in notes_qs:
plain_text = _html_to_plain_text(getattr(note, 'content', ''))
if plain_text:
collected.append(plain_text)
return collected
def is_smtp_available(timeout=5):
"""
Test if SMTP server is available before attempting to send emails.
Honors EMAIL_USE_SSL/EMAIL_USE_TLS to pick the right handshake.
Returns True if connection succeeds, False otherwise.
"""
email_host = getattr(settings, 'EMAIL_HOST', None)
email_port = getattr(settings, 'EMAIL_PORT', 25)
use_ssl = bool(getattr(settings, 'EMAIL_USE_SSL', False))
use_tls = bool(getattr(settings, 'EMAIL_USE_TLS', False))
# If no EMAIL_HOST is configured, SMTP is not available
if not email_host:
logger.warning("EMAIL_HOST not configured in settings")
return False
try:
email_port = int(email_port) if email_port else 25
except (ValueError, TypeError):
email_port = 25
# Heuristic: port 465 typically implies SSL even if EMAIL_USE_SSL isn't set
if email_port == 465:
use_ssl = True
try:
if use_ssl:
context = ssl.create_default_context()
with smtplib.SMTP_SSL(host=email_host, port=email_port, timeout=timeout, context=context) as smtp:
smtp.ehlo()
# No auth, just handshake
else:
with smtplib.SMTP(host=email_host, port=email_port, timeout=timeout) as smtp:
smtp.ehlo()
if use_tls:
context = ssl.create_default_context()
smtp.starttls(context=context)
smtp.ehlo()
return True
except (smtplib.SMTPException, socket.error, OSError, TimeoutError) as e:
logger.warning(f"SMTP server not available at {email_host}:{email_port} - {str(e)}")
return False
except Exception as e:
logger.error(f"Unexpected error testing SMTP connection: {str(e)}")
return False
def format_pause_description(reason_key: str, reason_other: str) -> str:
"""Return a human readable pause description using configured choices."""
label_obj = PAUSE_REASON_MAP.get(reason_key)
if not label_obj:
return _("Pause de l'intervention")
label = str(label_obj)
other_text = (reason_other or "").strip()
if reason_key == "other" and other_text:
return _("Pause de l'intervention - %(reason)s : %(details)s") % {
"reason": label,
"details": other_text,
}
return _("Pause de l'intervention - %(reason)s") % {"reason": label}
def _build_link_groups(intervention):
links = (
InterventionLink.objects
.filter(Q(first=intervention) | Q(second=intervention))
.select_related("first", "second")
.order_by("-created_at")
)
grouped = defaultdict(list)
for link in links:
try:
other = link.other(intervention)
except ValueError:
continue
grouped[link.link_type].append({
"intervention": other,
"link": link,
})
ordered_groups = []
for link_type, label in InterventionLink.LinkType.choices:
items = grouped.get(link_type)
if items:
ordered_groups.append({
"type": link_type,
"label": label,
"links": items,
})
return ordered_groups
def _group_operations_by_asset_and_template(operations, linked_assets):
"""Group intervention operations by asset (when linked) and template group."""
asset_lookup = {}
asset_content_type_lookup = {}
for intervention_asset in linked_assets:
if intervention_asset.content_type_id and intervention_asset.object_id:
asset_lookup[(intervention_asset.content_type_id, intervention_asset.object_id)] = getattr(intervention_asset, "asset", None)
asset_content_type_lookup[(intervention_asset.content_type_id, intervention_asset.object_id)] = intervention_asset.content_type
general_key = (None, None)
grouped = OrderedDict()
for operation in operations:
has_asset = bool(operation.asset_content_type_id and operation.asset_object_id)
asset_key = (operation.asset_content_type_id, operation.asset_object_id) if has_asset else general_key
entry = grouped.setdefault(asset_key, {
"asset": asset_lookup.get(asset_key),
"has_asset": asset_key != general_key,
"asset_content_type_id": asset_key[0],
"asset_object_id": asset_key[1],
"asset_content_type": asset_content_type_lookup.get(asset_key),
"template_groups": OrderedDict(),
})
if entry["asset"] is None and has_asset:
entry["asset"] = getattr(operation, "asset", None)
template_key = operation.template_group
entry["template_groups"].setdefault(template_key, []).append(operation)
def _is_operation_completed(op):
"""Vérifie si une opération est effectuée (valeur ou mesure indiquée)."""
if op.measurement_definition is not None:
vt = op.measurement_definition.value_type
if vt == 'duration':
return op.measured_duration is not None
elif vt == 'datetime':
return op.measured_datetime is not None
return op.measured_value is not None
return op.status is not None
def _get_operation_badge_color(op):
"""Returns 'success', 'warning', 'danger' or 'other' based on the status bootstrap_color or measurement range."""
if op.measurement_definition is not None:
vt = op.measurement_definition.value_type
if vt == 'duration':
if op.measured_duration is None:
return 'other'
return 'danger' if op.is_measurement_out_of_range() else 'success'
elif vt == 'datetime':
return 'success' if op.measured_datetime is not None else 'other'
# numeric
if op.measured_value is None:
return 'other'
return 'danger' if op.is_measurement_out_of_range() else 'success'
if op.status is not None and op.status.bootstrap_color:
color = op.status.bootstrap_color
if color == 'success':
return 'success'
elif color == 'warning':
return 'warning'
elif color == 'danger':
return 'danger'
return 'other'
def _serialize_group(data):
template_groups_list = []
for template_name, ops in data["template_groups"].items():
completed = sum(1 for op in ops if _is_operation_completed(op))
t_success = sum(1 for op in ops if _get_operation_badge_color(op) == 'success')
t_warning = sum(1 for op in ops if _get_operation_badge_color(op) == 'warning')
t_danger = sum(1 for op in ops if _get_operation_badge_color(op) == 'danger')
template_groups_list.append({
"name": template_name,
"operations": ops,
"completed_count": completed,
"pending_count": len(ops) - completed,
"success_count": t_success,
"warning_count": t_warning,
"danger_count": t_danger,
"other_count": len(ops) - t_success - t_warning - t_danger,
})
# Calculer le nombre total d'opérations pour ce groupe
all_operations = [op for tg in template_groups_list for op in tg["operations"]]
operations_count = len(all_operations)
completed_count = sum(1 for op in all_operations if _is_operation_completed(op))
pending_count = operations_count - completed_count
success_count = sum(1 for op in all_operations if _get_operation_badge_color(op) == 'success')
warning_count = sum(1 for op in all_operations if _get_operation_badge_color(op) == 'warning')
danger_count = sum(1 for op in all_operations if _get_operation_badge_color(op) == 'danger')
other_count = operations_count - success_count - warning_count - danger_count
# Get model name from content_type (safe access for templates)
asset_model_name = None
if data.get("asset_content_type"):
asset_model_name = data["asset_content_type"].model
return {
"asset": data["asset"],
"has_asset": data["has_asset"],
"asset_content_type_id": data.get("asset_content_type_id"),
"asset_object_id": data.get("asset_object_id"),
"asset_model_name": asset_model_name,
"template_groups": template_groups_list,
"operations_count": operations_count,
"completed_count": completed_count,
"pending_count": pending_count,
"success_count": success_count,
"warning_count": warning_count,
"danger_count": danger_count,
"other_count": other_count,
}
structured_groups = []
general_data = grouped.pop(general_key, None)
if general_data:
structured_groups.append(_serialize_group(general_data))
for data in grouped.values():
structured_groups.append(_serialize_group(data))
return structured_groups
def _build_replacement_model_options(asset_obj, category_id=None):
"""Return selectable model options for a given asset, optionally filtered by category."""
if not asset_obj or not hasattr(asset_obj, "_meta"):
return []
try:
model_field = asset_obj._meta.get_field("model")
except Exception:
return []
related_model = getattr(model_field, "related_model", None)
if not related_model:
return []
queryset = related_model.objects.all()
# Filter by category if provided
if category_id is not None:
queryset = queryset.filter(category_id=category_id)
if any(field.name == "code" for field in related_model._meta.fields):
queryset = queryset.order_by("code")
options = []
for item in queryset:
name_value = item.get_name() if hasattr(item, "get_name") else str(item)
category_name = item.category.get_name() if hasattr(item, 'category') and item.category else ""
options.append({
"id": item.id,
"label": name_value,
"category_id": item.category_id if hasattr(item, 'category') else None,
"category_name": category_name
})
options.sort(key=lambda x: x["label"] or "")
return options
def _get_asset_model_classes() -> List[type]:
global _ASSET_MODEL_CLASS_CACHE
if _ASSET_MODEL_CLASS_CACHE is None:
classes: List[type] = []
for model in apps.get_models():
if not isinstance(model, type):
continue
if not issubclass(model, AbstractAssetModel):
continue
if getattr(model._meta, "abstract", False):
continue
classes.append(model)
_ASSET_MODEL_CLASS_CACHE = classes
return _ASSET_MODEL_CLASS_CACHE
def _get_asset_classes() -> List[type]:
global _ASSET_CLASS_CACHE
if _ASSET_CLASS_CACHE is None:
classes: List[type] = []
for model in apps.get_models():
if not isinstance(model, type):
continue
if not issubclass(model, AbstractAsset):
continue
if getattr(model._meta, "abstract", False):
continue
classes.append(model)
_ASSET_CLASS_CACHE = classes
return _ASSET_CLASS_CACHE
def _get_allowed_asset_category_ids(intervention: Intervention) -> Set[int]:
if not getattr(intervention, "thematic", None):
return set()
return set(
AssetCategory.objects
.filter(thematic=intervention.thematic)
.values_list("id", flat=True)
)
def _parse_int_query_values(request, *param_names: str) -> List[int]:
raw_tokens: List[str] = []
for name in param_names:
raw_tokens.extend(request.GET.getlist(name))
single_value = request.GET.get(name)
if single_value:
raw_tokens.append(single_value)
parsed: List[int] = []
for token in raw_tokens:
if token is None:
continue
if isinstance(token, (list, tuple)):
candidates = token
else:
candidates = str(token).split(',')
for candidate in candidates:
value = str(candidate).strip()
if not value:
continue
try:
parsed.append(int(value))
except (TypeError, ValueError):
continue
deduplicated: List[int] = []
seen: Set[int] = set()
for value in parsed:
if value in seen:
continue
seen.add(value)
deduplicated.append(value)
return deduplicated
def _parse_model_filters(request) -> Dict[str, Set[int]]:
raw_tokens: List[str] = []
for name in ("model", "models"):
raw_tokens.extend(request.GET.getlist(name))
single_value = request.GET.get(name)
if single_value:
raw_tokens.append(single_value)
filters: Dict[str, Set[int]] = defaultdict(set)
for token in raw_tokens:
if token is None:
continue
if isinstance(token, (list, tuple)):
candidates = token
else:
candidates = str(token).split(',')
for candidate in candidates:
item = str(candidate).strip()
if not item or ':' not in item:
continue
label, object_id = item.split(':', 1)
label = label.strip().lower()
try:
pk = int(object_id.strip())
except (TypeError, ValueError):
continue
if not label:
continue
filters[label].add(pk)
return filters
def _parse_location_ct_obj_params(request, *param_names: str) -> List[tuple]:
"""Parse 'ct_id:obj_id' pairs from request GET params."""
raw_tokens: List[str] = []
for name in param_names:
raw_tokens.extend(request.GET.getlist(name))
pairs: List[tuple] = []
for token in raw_tokens:
if not token:
continue
for item in str(token).split(','):
item = item.strip()
if not item or ':' not in item:
continue
ct_str, obj_str = item.split(':', 1)
try:
pairs.append((int(ct_str.strip()), int(obj_str.strip())))
except (TypeError, ValueError):
continue
return pairs
def _format_location_label(location) -> str:
if not location:
return ""
code = getattr(location, 'code', '') or ''
name = location.get_name() if hasattr(location, 'get_name') else str(location)
name = name or ''
parts = [part for part in (code, name) if part]
return " – ".join(parts)
def _flatten_asset_collection(collection):
if collection is None:
return
if isinstance(collection, dict):
for value in collection.values():
yield from _flatten_asset_collection(value)
return
if isinstance(collection, (list, tuple, set)):
for value in collection:
yield from _flatten_asset_collection(value)
return
if isinstance(collection, QuerySet):
for item in collection.iterator():
yield item
return
yield collection
def _extract_assets_from_location(location):
getter = getattr(location, 'get_active_assets', None)
if not callable(getter):
return []
return _flatten_asset_collection(getter())
def _asset_matches_filters(asset, category_filter: Optional[Set[int]], model_filters: Dict[str, Set[int]], search_term: str) -> bool:
if category_filter and asset.category_id not in category_filter:
return False
if model_filters:
model_obj = getattr(asset, 'model', None)
if not model_obj:
return False
model_label = model_obj._meta.model_name.lower()
allowed_ids = model_filters.get(model_label)
if not allowed_ids or model_obj.pk not in allowed_ids:
return False
if search_term:
asset_name = asset.get_name() if hasattr(asset, 'get_name') else str(asset)
haystack = f"{getattr(asset, 'code', '')} {asset_name or ''}".lower()
if search_term not in haystack:
return False
return True
def _serialize_asset_entry(asset, ct: ContentType, location_labels: Set[str]):
category = getattr(asset, 'category', None)
category_name = category.get_name() if category else None
category_code = getattr(category, 'code', None) if category else None
model_obj = getattr(asset, 'model', None)
model_key = None
model_display = None
if model_obj:
model_label = model_obj.get_name() if hasattr(model_obj, 'get_name') else str(model_obj)
model_code = getattr(model_obj, 'code', None)
if model_code and model_label:
model_display = f"{model_code} – {model_label}"
else:
model_display = model_label or model_code
model_key = f"{model_obj._meta.model_name}:{model_obj.pk}"
asset_name = asset.get_name() if hasattr(asset, 'get_name') else str(asset)
display_label = asset_name or getattr(asset, 'code', '')
detail_url = asset.get_absolute_url() if hasattr(asset, 'get_absolute_url') else None
return {
"content_type_id": ct.id,
"content_type_model": ct.model,
"object_id": asset.pk,
"code": getattr(asset, 'code', None),
"name": asset_name,
"display_label": display_label,
"category_id": category.id if category else None,
"category_name": category_name,
"category_code": category_code,
"model_key": model_key,
"model_name": model_display,
"model_id": model_obj.pk if model_obj else None,
"model_label": model_obj._meta.model_name if model_obj else None,
"location_labels": sorted(label for label in location_labels if label),
"detail_url": detail_url,
}
def _collect_assets_from_locations(intervention: Intervention, locations, category_ids: Optional[Set[int]], model_filters: Dict[str, Set[int]], search_term: str):
existing_pairs = set(
intervention.interventionasset_set.values_list('content_type_id', 'object_id')
)
ct_cache: Dict[type, ContentType] = {}
selected: List[tuple] = []
location_map: Dict[tuple, Set[str]] = {}
seen: Set[tuple] = set()
match_count = 0
max_matches = MAX_MOBILE_ASSET_RESULTS + 1
for link in locations:
location = getattr(link, 'location', None)
if not location:
continue
loc_label = _format_location_label(location)
for asset in _extract_assets_from_location(location):
if not isinstance(asset, AbstractAsset):
continue
ct = ct_cache.setdefault(asset.__class__, ContentType.objects.get_for_model(asset.__class__))
key = (ct.id, asset.pk)
if key in existing_pairs:
continue
if key not in seen:
if not _asset_matches_filters(asset, category_ids, model_filters, search_term):
continue
seen.add(key)
match_count += 1
if len(selected) < MAX_MOBILE_ASSET_RESULTS:
selected.append((asset, ct, key))
location_map.setdefault(key, set()).add(loc_label)
if match_count >= max_matches:
break
else:
if key in location_map:
location_map[key].add(loc_label)
if match_count >= max_matches:
break
assets_payload = [
_serialize_asset_entry(asset, ct, location_map.get(key, set()))
for asset, ct, key in selected
]
return {
"assets": assets_payload,
"match_count": match_count,
"has_more": match_count > MAX_MOBILE_ASSET_RESULTS,
}
def _collect_assets_from_categories(intervention: Intervention, category_ids: Set[int], model_filters: Dict[str, Set[int]], search_term: str):
existing_pairs = set(
intervention.interventionasset_set.values_list('content_type_id', 'object_id')
)
ct_cache: Dict[type, ContentType] = {}
selected: List[tuple] = []
seen: Set[tuple] = set()
match_count = 0
max_matches = MAX_MOBILE_ASSET_RESULTS + 1
for Model in _get_asset_classes():
queryset = Model.objects.filter(category_id__in=category_ids).order_by('code')
for asset in queryset.iterator():
ct = ct_cache.setdefault(Model, ContentType.objects.get_for_model(Model))
key = (ct.id, asset.pk)
if key in existing_pairs or key in seen:
continue
if not _asset_matches_filters(asset, None, model_filters, search_term):
continue
seen.add(key)
match_count += 1
if len(selected) < MAX_MOBILE_ASSET_RESULTS:
selected.append((asset, ct, key))
if match_count >= max_matches:
break
if match_count >= max_matches:
break
assets_payload = [
_serialize_asset_entry(asset, ct, set())
for asset, ct, _ in selected
]
return {
"assets": assets_payload,
"match_count": match_count,
"has_more": match_count > MAX_MOBILE_ASSET_RESULTS,
}
def _resolve_location_pairs_to_links(ct_obj_pairs: List[tuple]):
"""
Résout des paires (ct_id, obj_id) en objets simples avec attribut .location,
compatibles avec _collect_assets_from_locations.
"""
from types import SimpleNamespace
reg_ct: Dict[int, type] = {}
for meta in LOCATION_REGISTRY.values():
ct = ContentType.objects.get_for_model(meta['model'])
reg_ct[ct.id] = meta['model']
grouped: Dict[int, List[int]] = defaultdict(list)
for ct_id, obj_id in ct_obj_pairs:
if ct_id in reg_ct:
grouped[ct_id].append(obj_id)
links = []
for ct_id, obj_ids in grouped.items():
Model = reg_ct[ct_id]
for loc in Model.objects.filter(pk__in=obj_ids):
links.append(SimpleNamespace(location=loc, content_type_id=ct_id, object_id=loc.pk))
return links
def _search_mobile_assets(intervention: Intervention, locations, category_ids: Optional[Set[int]], model_filters: Dict[str, Set[int]], search_term: str):
has_locations = bool(locations)
if has_locations:
result = _collect_assets_from_locations(intervention, locations, category_ids, model_filters, search_term)
else:
if not category_ids:
return {
"assets": [],
"match_count": 0,
"has_more": False,
"has_locations": False,
"requires_category": True,
}
result = _collect_assets_from_categories(intervention, category_ids, model_filters, search_term)
result.update({
"has_locations": has_locations,
"requires_category": not has_locations and not category_ids,
})
return result
def _attach_equipment_templates(intervention, templates):
if templates:
intervention.equipment_operation_templates.add(*templates)
def _assign_guided_template(intervention, templates):
"""
If any selected OperationTemplate has a guided_template FK set,
auto-assign the first one found to the intervention.
"""
if not templates:
return
for tmpl in templates:
if tmpl.guided_template_id:
intervention.guided_operation_template = tmpl.guided_template
intervention.save(update_fields=['guided_operation_template'])
return
def _assign_team_from_templates(intervention, templates):
"""
If any selected OperationTemplate has a team FK set,
auto-assign the first one found to the intervention.
"""
if not templates:
return
for tmpl in templates:
if tmpl.team_id:
intervention.assigned_team = tmpl.team
fields_to_update = ['assigned_team']
if not intervention.assigned_provider_id and tmpl.team.company_id:
intervention.assigned_provider = tmpl.team.company
fields_to_update.append('assigned_provider')
intervention.save(update_fields=fields_to_update)
return
def _assign_contract_posts_from_templates(intervention, templates):
"""
Associe les postes de contrat par défaut configurés pour les OperationTemplates donnés
sur le contrat de l'intervention.
Si plusieurs templates ont des postes en doublon, le premier trouvé est conservé.
"""
if not templates:
return
contract = intervention.contract or (intervention.order.contract if intervention.order else None)
if not contract:
return
existing_post_ids = set(
intervention.interv_contract_posts.values_list('contract_post_id', flat=True)
)
posts_to_create = []
for template in templates:
template_posts = (
ContractOperationTemplatePost.objects
.filter(contract=contract, operation_template=template)
.select_related('contract_post')
.order_by('contract_post__order_number', 'id')
)
for tp in template_posts:
if tp.contract_post_id not in existing_post_ids:
posts_to_create.append(
InterventionContractPost(
intervention=intervention,
contract_post=tp.contract_post,
quantity=tp.default_quantity,
comment=tp.comment or '',
)
)
existing_post_ids.add(tp.contract_post_id)
if posts_to_create:
InterventionContractPost.objects.bulk_create(posts_to_create)
def _get_equipment_templates_for_intervention(intervention):
templates = list(intervention.equipment_operation_templates.all())
if templates:
return templates
template_names = (
intervention.operations
.filter(asset_content_type__isnull=False, template_group__isnull=False)
.values_list('template_group', flat=True)
.distinct()
)
if not template_names:
return []
inferred = list(
OperationTemplate.objects
.filter(name__in=template_names)
)
if inferred:
intervention.equipment_operation_templates.add(*inferred)
return inferred
def _add_operations_from_templates(intervention, templates_list, asset_obj=None, asset_ct=None, only_generic=None):
"""
Ajoute des opérations à une intervention à partir de templates.
Args:
intervention: L'intervention à laquelle ajouter les opérations
templates_list: Liste de templates d'opérations
asset_obj: L'objet asset (si opérations liées à un équipement)
asset_ct: Le ContentType de l'asset
only_generic: Si True, n'ajoute que les items génériques. Si False, n'ajoute que les items non-génériques.
Si None, ajoute tous les items en respectant leur attribut is_generic.
"""
if not templates_list:
return []
# Pour les opérations non-génériques liées à un asset, vérifier les doublons
existing_template_groups_for_asset = set()
if asset_obj is not None and asset_ct is not None:
existing_template_groups_for_asset = set(
Operation.objects
.filter(
intervention=intervention,
asset_content_type=asset_ct,
asset_object_id=asset_obj.pk,
is_generic=False,
)
.values_list('template_group', flat=True)
)
# Pour les opérations génériques, vérifier les doublons (sans asset)
existing_generic_template_groups = set(
Operation.objects
.filter(
intervention=intervention,
is_generic=True,
)
.values_list('template_group', flat=True)
)
operations_to_create = []
for template in templates_list:
items = OperationTemplateItem.objects.filter(template=template).order_by('order')
for item in items:
# Filtrer selon only_generic si spécifié
if only_generic is True and not item.is_generic:
continue
if only_generic is False and item.is_generic:
continue
# Vérifier les doublons
if item.is_generic:
# Pour les items génériques, vérifier si déjà créé pour ce template
if template.name in existing_generic_template_groups:
# Vérifier si cet item spécifique existe déjà
if Operation.objects.filter(
intervention=intervention,
template_group=template.name,
name_fr=item.name_fr,
is_generic=True
).exists():
continue
else:
# Pour les items non-génériques, vérifier si déjà créé pour cet asset
if asset_obj and asset_ct and template.name in existing_template_groups_for_asset:
if Operation.objects.filter(
intervention=intervention,
template_group=template.name,
name_fr=item.name_fr,
asset_content_type=asset_ct,
asset_object_id=asset_obj.pk,
is_generic=False
).exists():
continue
op = Operation(
intervention=intervention,
name_fr=item.name_fr,
name_nl=item.name_nl,
status_set=item.status_set,
measurement_definition=item.measurement_definition,
order=item.order,
template_group=template.name,
requires_photo=item.requires_photo,
is_generic=item.is_generic,
)
# Pour les items non-génériques, lier à l'asset si fourni
if not item.is_generic and asset_obj and asset_ct:
op.asset_content_type = asset_ct
op.asset_object_id = asset_obj.pk
operations_to_create.append(op)
if not operations_to_create:
return []
return list(Operation.objects.bulk_create(operations_to_create))
def _resolve_assets_from_pairs(pairs):
assets = []
if not pairs:
return assets
by_ct = defaultdict(set)
for ct_id, obj_id in pairs:
by_ct[ct_id].add(obj_id)
ct_map = {ct.id: ct for ct in ContentType.objects.filter(id__in=by_ct.keys())}
for ct_id, obj_ids in by_ct.items():
ct = ct_map.get(ct_id)
if not ct:
continue
Model = ct.model_class()
if not Model:
continue
obj_map = {obj.pk: obj for obj in Model.objects.filter(pk__in=obj_ids)}
for obj_id in obj_ids:
asset_obj = obj_map.get(obj_id)
if asset_obj:
assets.append((ct, asset_obj))
return assets
def _asset_matches_template_filters(asset_obj, template, allow_unfiltered=True):
"""
Check if an asset matches the template's category and asset model filters.
If no filters are defined, behavior depends on allow_unfiltered:
- True: treat as match (used when user selected assets explicitly)
- False: treat as no match (used when iterating child assets of locations)
"""
category_ids = template.get_category_ids()
model_filters = template.get_asset_model_filters()
# No filters defined -> follow caller preference
if not category_ids and not model_filters:
return allow_unfiltered
# Check category filter
if category_ids:
asset_category = getattr(asset_obj, 'category', None)
if not asset_category or asset_category.id not in category_ids:
return False
# Check asset model filter
if model_filters:
asset_model = getattr(asset_obj, 'model', None)
if not asset_model:
return False
model_name = asset_model._meta.model_name.lower()
allowed_ids = model_filters.get(model_name)
if not allowed_ids or asset_model.pk not in allowed_ids:
return False
return True
def _get_templates_with_equipment_filters(templates):
"""
Filter templates that have category or model filters defined.
Returns only templates that should match child assets from locations.
"""
return [t for t in templates if t.get_category_ids() or t.get_asset_model_filters()]
def _create_equipment_operations_for_assets(intervention, assets_with_ct, ignore_filters=False):
"""
Create operations for assets based on equipment templates.
Args:
intervention: The intervention
assets_with_ct: List of (content_type, asset_object) tuples
ignore_filters: If True, ignore category/model filters and apply all templates.
Use this when user manually adds an asset (they chose it explicitly).
"""
templates = _get_equipment_templates_for_intervention(intervention)
if not templates or not assets_with_ct:
return 0
total_created = 0
for asset_ct, asset_obj in assets_with_ct:
if ignore_filters:
# User manually added this asset, apply all templates regardless of filters
applicable_templates = templates
else:
# Automatic linking (e.g., from location children), respect filters
applicable_templates = [
t for t in templates
if _asset_matches_template_filters(asset_obj, t)
]
if not applicable_templates:
continue
created_ops = _add_operations_from_templates(
intervention,
applicable_templates,
asset_obj=asset_obj,
asset_ct=asset_ct,
)
total_created += len(created_ops)
return total_created
def _unlink_operations_from_assets(intervention, asset_pairs):
"""
Unlink operations from assets by setting their asset references to NULL,
effectively converting them to general operations.
Only processes operations without values.
"""
if not asset_pairs:
return 0, 0
value_filter = (
Q(measured_value__isnull=False)
| Q(measured_duration__isnull=False)
| Q(measured_datetime__isnull=False)
| Q(status__isnull=False)
| (Q(comment__isnull=False) & ~Q(comment__exact=""))
)
unlinked_total = 0
assets_with_values = 0
for ct_id, obj_id in asset_pairs:
operations_qs = Operation.objects.filter(
intervention=intervention,
asset_content_type_id=ct_id,
asset_object_id=obj_id,
)
if not operations_qs.exists():
continue
if operations_qs.filter(value_filter).exists():
assets_with_values += 1
continue
# Unlink operations by setting asset references to NULL
unlinked_total += operations_qs.update(
asset_content_type=None,
asset_object_id=None
)
return unlinked_total, assets_with_values
def _force_delete_operations_for_assets(intervention, asset_pairs):
"""
Force delete all operations for given assets, regardless of whether they have values.
"""
if not asset_pairs:
return 0, 0
deleted_total = 0
for ct_id, obj_id in asset_pairs:
operations_qs = Operation.objects.filter(
intervention=intervention,
asset_content_type_id=ct_id,
asset_object_id=obj_id,
)
deleted_total += operations_qs.delete()[0]
return deleted_total, 0
def _mark_asset_as_removed(asset_key, user):
"""
Mark an asset as removed by setting marked_as_removed_date and marked_as_removed_by fields.
The asset_key is in format "content_type_id:object_id"
"""
if not asset_key or ':' not in asset_key:
return
try:
ct_id, obj_id = asset_key.split(':')
ct_id = int(ct_id)
obj_id = int(obj_id)
except (ValueError, TypeError):
return
try:
from django.utils import timezone
ct = ContentType.objects.get(pk=ct_id)
asset_model = ct.model_class()
if asset_model:
asset = asset_model.objects.filter(pk=obj_id).first()
if asset and hasattr(asset, 'marked_as_removed_date'):
asset.marked_as_removed_date = timezone.now()
if user and hasattr(asset, 'marked_as_removed_by'):
asset.marked_as_removed_by = user
asset.save(update_fields=['marked_as_removed_date', 'marked_as_removed_by'] if hasattr(asset, 'marked_as_removed_by') else ['marked_as_removed_date'])
except Exception as e:
logger.exception(f"Error marking asset as removed: {e}")
def _delete_operations_for_assets(intervention, asset_pairs):
if not asset_pairs:
return 0, 0
value_filter = (
Q(measured_value__isnull=False)
| Q(measured_duration__isnull=False)
| Q(measured_datetime__isnull=False)
| Q(status__isnull=False)
| (Q(comment__isnull=False) & ~Q(comment__exact=""))
)
deleted_total = 0
assets_with_values = 0
for ct_id, obj_id in asset_pairs:
operations_qs = Operation.objects.filter(
intervention=intervention,
asset_content_type_id=ct_id,
asset_object_id=obj_id,
)
if not operations_qs.exists():
continue
if operations_qs.filter(value_filter).exists():
assets_with_values += 1
continue
deleted_total += operations_qs.delete()[0]
return deleted_total, assets_with_values
def _sync_intervention_assets_data(
intervention: Intervention,
normalized_assets: Optional[List[Dict[str, int]]],
delete_operations: bool = False,
mark_as_removed: bool = False,
removed_asset_key: Optional[str] = None,
removed_asset_keys: Optional[List[str]] = None,
user = None,
request = None,
) -> Dict[str, int]:
normalized_assets = normalized_assets or []
prepared_assets = []
for entry in normalized_assets:
if not isinstance(entry, dict):
continue
model_label = str(entry.get('model', '')).strip().lower()
if not model_label:
continue
try:
object_id = int(entry.get('object_id'))
except (TypeError, ValueError):
continue
prepared_assets.append((model_label, object_id))
if not prepared_assets:
target_pairs: Set[Tuple[int, int]] = set()
model_names: List[str] = []
else:
model_names = sorted({label for label, _ in prepared_assets})
target_pairs = set()
cts = {
ct.model: ct
for ct in ContentType.objects.filter(app_label='assets', model__in=model_names)
}
missing_models = [model for model in model_names if model not in cts]
if missing_models:
readable = ', '.join(missing_models)
raise ValueError(_("Modèles inconnus: %(models)s") % {"models": readable})
for model_label, object_id in prepared_assets:
ct = cts.get(model_label)
if not ct:
continue
target_pairs.add((ct.id, object_id))
existing_qs = InterventionAsset.objects.filter(intervention=intervention)
existing_pairs = set(existing_qs.values_list('content_type_id', 'object_id'))
to_add = target_pairs - existing_pairs
to_remove = existing_pairs - target_pairs
created_count = 0
deleted_count = 0
operations_deleted = 0
assets_with_values = 0
with transaction.atomic():
if to_remove:
q = Q()
for ct_id, obj_id in to_remove:
q |= Q(content_type_id=ct_id, object_id=obj_id)
if q:
deletion_result = existing_qs.filter(q)
deleted_count = deletion_result.delete()[0]
# Handle operations deletion based on delete_operations flag
if delete_operations:
# Force delete operations regardless of values
operations_deleted, assets_with_values = _force_delete_operations_for_assets(
intervention,
to_remove,
)
else:
# Keep operations
pass
# Mark asset(s) as removed if requested
if mark_as_removed:
# Support both single key and list of keys
keys_to_mark = list(removed_asset_keys) if removed_asset_keys else []
if removed_asset_key and removed_asset_key not in keys_to_mark:
keys_to_mark.append(removed_asset_key)
for key in keys_to_mark:
_mark_asset_as_removed(key, user)
if to_add:
new_relations = [
InterventionAsset(
intervention=intervention,
content_type_id=ct_id,
object_id=obj_id,
)
for ct_id, obj_id in to_add
]
InterventionAsset.objects.bulk_create(new_relations)
created_count = len(new_relations)
operations_created = 0
if to_add:
assets_with_ct = _resolve_assets_from_pairs(to_add)
# For preventive interventions, store newly linked assets in session for template selection
# For corrective interventions, create operations automatically
if intervention.maintain_type == 'preventive' and request:
# Store newly linked asset IDs in session for template selection modal
new_asset_keys = [f"{ct_id}:{obj_id}" for ct_id, obj_id in to_add]
session_key = f'new_linked_assets_{intervention.id}'
request.session[session_key] = new_asset_keys
request.session.modified = True
else:
# Corrective or no request: create operations automatically
operations_created = _create_equipment_operations_for_assets(intervention, assets_with_ct)
return {
"created": created_count,
"deleted": deleted_count,
"operations_created": operations_created,
"operations_deleted": operations_deleted,
"assets_with_values": assets_with_values,
"target_total": len(target_pairs),
"requested_assets": len(prepared_assets),
}
def _expand_csv_params(param_list):
"""Expands comma-separated values in a list of parameters."""
if not param_list:
return None
result = []
for item in param_list:
result.extend([v.strip() for v in item.split(',') if v.strip()])
return result if result else None
def annotate_preparation_status(qs):
"""
Annote le queryset avec un champ 'prep_state' indiquant l'état du bon de préparation.
Valeurs: 'processed' (tous traités/annulés), 'not_processed' (au moins un non traité), 'none' (pas de bon).
"""
from stock.models import PreparationOrder
# Compte les bons non traités (à traiter ou en attente)
untreated_count = Count(
'preparation_orders',
filter=Q(preparation_orders__status__in=['to_process', 'pending'])
)
# Compte le total des bons
total_count = Count('preparation_orders')
# Annote le queryset
qs = qs.annotate(
_prep_untreated_count=untreated_count,
_prep_total_count=total_count
)
# Ajoute le champ 'prep_state' basé sur les annotations
qs = qs.annotate(
prep_state=Case(
When(_prep_total_count=0, then=Value('none')),
When(_prep_untreated_count__gt=0, then=Value('not_processed')),
default=Value('processed'),
output_field=CharField()
)
)
return qs
def apply_intervention_filters_from_querydict(qs, params, user=None):
"""
Applique les filtres UI standard sur un queryset d'interventions.
`params` est un QueryDict (request.GET) ou objet compatible (.getlist / .get).
Ne fait pas d'annotation GeoJSON – uniquement les filtres métier.
Retourne le queryset filtré (sans ordering ni limit).
"""
thematics = params.getlist('thematic') or None
locations = params.get('locations') or None
statuses = _expand_csv_params(params.getlist('status'))
priorities = _expand_csv_params(params.getlist('priority'))
maintain_types = _expand_csv_params(params.getlist('maintain_type'))
categories = _expand_csv_params(params.getlist('category'))
types = _expand_csv_params(params.getlist('type'))
providers = _expand_csv_params(params.getlist('provider'))
contracts = _expand_csv_params(params.getlist('contract'))
assigned_teams = _expand_csv_params(params.getlist('assigned_team'))
assigned_members = _expand_csv_params(params.getlist('assigned_member'))
intervention_managers = _expand_csv_params(params.getlist('intervention_manager'))
pause_reasons = _expand_csv_params(params.getlist('pause_reason'))
preparation_states = _expand_csv_params(params.getlist('preparation_state'))
begin_date_min = params.get('begin_date_min') or None
begin_date_max = params.get('begin_date_max') or None
end_date_min = params.get('end_date_min') or None
end_date_max = params.get('end_date_max') or None
date_type = params.get('date_type', 'all')
overdue = params.get('overdue')
followed_only = params.get('followed_only')
to_be_planned_soon = params.get('to_be_planned_soon')
is_checked_filter = params.get('is_checked')
search_q = (params.get('q') or '').strip()
search_mode = (params.get('search_mode') or 'and').lower()
# Annote le queryset avec le statut du bon de préparation
qs = annotate_preparation_status(qs)
if thematics:
qs = qs.filter(thematic__code__in=thematics)
if locations:
try:
selected_loc = json.loads(locations) if isinstance(locations, str) else locations
except Exception:
selected_loc = []
allowed_ct_ids = _allowed_ct_ids_from_thematics(thematics) if thematics else None
qs = apply_location_filter(qs, selected_loc, allowed_ct_ids=allowed_ct_ids)
if statuses:
qs = qs.filter(status__in=statuses)
if pause_reasons and statuses and set(statuses) == {'on_pause'}:
qs = qs.filter(pause_reason__in=pause_reasons)
if priorities:
qs = qs.filter(priority__in=priorities)
if maintain_types:
qs = qs.filter(maintain_type__in=maintain_types)
if categories:
qs = qs.filter(asset_category_id__in=categories)
if types:
qs = qs.filter(type__in=types)
if providers:
qs = qs.filter(assigned_provider_id__in=providers)
if contracts:
qs = qs.filter(contract_id__in=contracts)
if assigned_members:
qs = qs.filter(assigned_member_id__in=assigned_members)
if assigned_teams:
qs = qs.filter(Q(assigned_team_id__in=assigned_teams) | Q(assigned_member__teams__in=assigned_teams)).distinct()
if intervention_managers:
qs = qs.filter(intervention_manager_id__in=intervention_managers)
if followed_only == '1' and user and user.is_authenticated:
qs = qs.filter(subscriptions__user=user)
# Date filters (same logic as interventions_geojson but without effective_begin annotation)
if date_type == 'planned':
eff_begin = Coalesce('begin_time', 'planned_begin_time', output_field=DateTimeField())
eff_end = Coalesce('end_time', 'planned_end_time', output_field=DateTimeField())
else:
eff_begin = Coalesce('begin_time', 'planned_begin_time', 'expected_begin_time', output_field=DateTimeField())
eff_end = Coalesce('end_time', 'planned_end_time', 'expected_end_time', output_field=DateTimeField())
if begin_date_min or begin_date_max or end_date_min or end_date_max:
qs = qs.annotate(_eff_begin=eff_begin, _eff_end=eff_end)
if begin_date_min:
dt = make_aware(datetime.strptime(begin_date_min, '%Y-%m-%d'), get_current_timezone())
qs = qs.filter(_eff_begin__gte=dt)
if begin_date_max:
dt = datetime.strptime(begin_date_max, '%Y-%m-%d').replace(hour=23, minute=59, second=59, microsecond=999999)
dt = make_aware(dt, get_current_timezone())
qs = qs.filter(_eff_begin__lte=dt)
if end_date_min:
dt = make_aware(datetime.strptime(end_date_min, '%Y-%m-%d'), get_current_timezone())
qs = qs.filter(_eff_end__gte=dt)
if end_date_max:
dt = datetime.strptime(end_date_max, '%Y-%m-%d').replace(hour=23, minute=59, second=59, microsecond=999999)
dt = make_aware(dt, get_current_timezone())
qs = qs.filter(_eff_end__lte=dt)
if overdue == '1':
_now = django_timezone.now()
_1_day_ago = _now - timedelta(days=1)
_7d = _now - timedelta(days=7)
_latest_in_progress_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='in_progress',
).order_by('-event_time').values('event_time')[:1]
_latest_assigned_sq = InterventionTimeLine.objects.filter(
intervention=OuterRef('pk'),
to_status='assigned',
).order_by('-event_time').values('event_time')[:1]
_eff_begin_nt = Coalesce('planned_begin_time', 'expected_begin_time', output_field=DateTimeField())
qs = qs.annotate(
_nt_eff_begin=_eff_begin_nt,
_nt_last_in_progress=Coalesce(Subquery(_latest_in_progress_sq), 'begin_time', output_field=DateTimeField()),
_nt_last_assigned=Coalesce(Subquery(_latest_assigned_sq), 'transmission_time', 'creation_time', output_field=DateTimeField()),
).filter(
Q(status__in=['to_be_processed', 'assigned'],
_nt_eff_begin__isnull=False, _nt_eff_begin__lt=_1_day_ago)
| Q(status='in_progress', _nt_last_in_progress__isnull=False, _nt_last_in_progress__lt=_7d)
| Q(status='assigned', _nt_last_assigned__isnull=False, _nt_last_assigned__lt=_7d)
)
if to_be_planned_soon == '1':
three_months = django_timezone.now() + timedelta(days=90)
qs = qs.filter(Q(expected_begin_time__isnull=True) | Q(expected_begin_time__lt=three_months))
if is_checked_filter == 'true':
qs = qs.filter(is_checked=True)
elif is_checked_filter == 'false':
qs = qs.filter(is_checked=False)
if preparation_states:
qs = qs.filter(prep_state__in=preparation_states)
if search_q:
if search_mode not in ('and', 'or'):
search_mode = 'and'
words = search_q.split()
from common.hooks import hooks
qs = hooks.apply_search_annotations(qs)
def _word_q(word):
q_base = (
Q(title__icontains=word)
| Q(code__icontains=word)
| Q(location_code__icontains=word)
| Q(contract__contract_number__icontains=word)
)
q_base |= hooks.get_search_filters(word)
return q_base
combined = Q()
if search_mode == 'and':
for w in words:
combined &= _word_q(w)
else:
for w in words:
combined |= _word_q(w)
qs = qs.filter(combined)
return qs
def _parse_selected_locations(request):
raw = request.GET.get("locations")
try:
return json.loads(raw) if raw else []
except Exception:
return []
def _allowed_ct_ids_from_thematics(thematics):
# optionnel: si vous voulez verrouiller par thèmes côté serveur
from assets.views.locations import registry_items_for_thematics, content_type_for_model
cts = set()
for key, meta in registry_items_for_thematics(thematics):
cts.add(content_type_for_model(meta["model"]).id)
return cts
def _expand_hierarchical_location_ids(ct_id: int, obj_ids: list[int]) -> list[int]:
"""
Si le modèle visé par ct_id possède un champ 'parent_location',
on inclut récursivement tous les enfants des ids fournis.
"""
try:
Model = ContentType.objects.get_for_id(ct_id).model_class()
except ContentType.DoesNotExist:
return obj_ids
# Détecter un modèle hiérarchique (StructureLocation, …)
try:
Model._meta.get_field('parent_location')
except FieldDoesNotExist:
return obj_ids # non hiérarchique → on ne change rien
all_ids = set(map(int, obj_ids))
frontier = set(all_ids)
while frontier:
children = set(
Model.objects
.filter(parent_location_id__in=frontier)
.values_list('id', flat=True)
)
new = children - all_ids
if not new:
break
all_ids |= new
frontier = new
return list(all_ids)
def apply_location_filter(qs, selected, allowed_ct_ids=None):
"""
selected: [{"ct": <content_type_id>, "obj": <int>}, ...]
"""
if not selected:
return qs
pairs = [(s.get("ct"), s.get("obj")) for s in selected if s.get("ct") and s.get("obj")]
if allowed_ct_ids:
pairs = [(ct, obj) for (ct, obj) in pairs if ct in allowed_ct_ids]
if not pairs:
return qs.none()
by_ct = {}
for ct, obj in pairs:
by_ct.setdefault(ct, set()).add(int(obj))
q_total = Q()
for ct, objs in by_ct.items():
# inclusion des enfants si hiérarchie présente (ex. StructureLocation)
expanded_objs = _expand_hierarchical_location_ids(ct, list(objs))
q_total |= Q(locations__content_type_id=ct, locations__object_id__in=expanded_objs)
# Optionnel (si risque de doublons via plusieurs localisations liées) :
# return qs.filter(q_total).distinct()
return qs.filter(q_total)
def _registry_by_ct():
"""Map ContentType.id -> meta (model, label, theme, type_key, ct)."""
out = {}
for type_key, meta in LOCATION_REGISTRY.items():
ct = ContentType.objects.get_for_model(meta["model"])
out[ct.id] = {**meta, "type_key": type_key, "ct": ct.id}
return out
def _resolve_locations_payload(param_json: str | None):
"""
param_json: string JSON '[{"ct": 29, "obj": 123}, ...]'
Retourne une liste d'items UI-consistents avec l'autocomplete:
{ct,obj,type,type_label,thematic,name,code,label,lon,lat}
"""
if not param_json:
return []
try:
raw = json.loads(param_json)
except Exception:
return []
pairs = [(int(x["ct"]), int(x["obj"])) for x in raw if isinstance(x, dict) and "ct" in x and "obj" in x]
if not pairs:
return []
# dédoublonnage par ct
grouped = {}
for ct_id, obj in pairs:
grouped.setdefault(ct_id, set()).add(obj)
reg_ct = _registry_by_ct()
lang = (get_language() or "fr").split("-")[0]
is_nl = (lang == "nl")
results = []
for ct_id, ids in grouped.items():
meta = reg_ct.get(ct_id)
if not meta:
continue
Model = meta["model"]
# On récupère juste ce qu'il faut pour construire le label
for r in Model.objects.filter(pk__in=list(ids)).values("id", "code", "name_fr", "name_nl", "lon", "lat"):
name = r["name_nl"] if is_nl else r["name_fr"]
if not name:
name = r["name_fr"] or r["name_nl"] or ""
label = f"{name} ({r['code']})"
results.append({
"ct": ct_id,
"obj": r["id"],
"type": meta["type_key"],
"type_label": meta["label"],
"thematic": meta.get("theme"),
"name": name,
"code": r["code"],
"label": label,
"lon": r["lon"],
"lat": r["lat"],
})
return results
def _build_guided_media_json(guided_op_data):
"""Serialize existing GuidedOperationMedia for the edit-mode JS."""
if not guided_op_data:
return []
from common.private_files.helpers import build_private_url
result = []
for media in guided_op_data.media.all():
result.append({
'media_id': media.pk,
'field_key': media.field_key,
'media_type': media.media_type,
'description': media.description,
'file_url': build_private_url('interventions', 'guidedoperationmedia', media.pk, 'file'),
})
return result
def update_operation_comment(request, operation_id):
if not request.user.is_authenticated:
return JsonResponse({'success': False, 'error': _('Authentification requise.')}, status=401)
operation = get_object_or_404(
Operation.objects.select_related('intervention'),
pk=operation_id
)
user_config = (UserConfig.objects
.select_related('user')
.prefetch_related('roles')
.filter(user=request.user)
.first())
if not user_config:
return JsonResponse({'success': False, 'error': _('Vous n\'avez pas accès à cette intervention.')}, status=403)
allowed_update_fields = get_allowed_update_fields(request.user, operation.intervention)
if 'operations' not in allowed_update_fields and not can_edit_intervention(request.user, operation.intervention, user_config=user_config):
return JsonResponse({'success': False, 'error': _('Vous n\'avez pas la permission de modifier cette opération.')}, status=403)
comment_value = ''
if request.content_type == 'application/json':
try:
payload = json.loads(request.body.decode('utf-8') or '{}')
except json.JSONDecodeError:
payload = {}
comment_value = payload.get('comment', '')
else:
comment_value = request.POST.get('comment', '')
comment_value = (comment_value or '').strip()
operation.comment = comment_value
operation.save(update_fields=['comment'])
return JsonResponse({'success': True, 'comment': comment_value})
def _log_planning_change(intervention, user, previous_begin, previous_end, previous_status=None):
changes = []
fmt = lambda dt: dj_format_date(localtime(dt), 'd/m/Y H:i') if dt else _('Non défini')
if previous_begin != intervention.planned_begin_time:
changes.append(_('Début planifié : %(old)s → %(new)s') % {
'old': fmt(previous_begin),
'new': fmt(intervention.planned_begin_time),
})
if previous_end != intervention.planned_end_time:
changes.append(_('Fin planifiée : %(old)s → %(new)s') % {
'old': fmt(previous_end),
'new': fmt(intervention.planned_end_time),
})
if previous_status is not None and previous_status != intervention.status:
from interventions.models import STATUS_CHOICES
old_label = STATUS_CHOICES.get(previous_status, previous_status)
new_label = STATUS_CHOICES.get(intervention.status, intervention.status)
changes.append(_('Statut : %(old)s → %(new)s') % {
'old': old_label,
'new': new_label,
})
if not changes:
return
InterventionPlanificationTimeLine.objects.create(
intervention=intervention,
event_user=user,
event_time=timezone.now(),
event_type='planned_begin',
event_description=' | '.join(changes)
)
def _parse_pairs_json(raw, key_ct='ct', key_obj='obj'):
"""Parse '[{"ct": 29, "obj": 123}, ...]' -> [(29, 123), ...] sans crash."""
if not raw:
return []
try:
arr = json.loads(raw)
out = []
for item in arr if isinstance(arr, list) else []:
ct = item.get(key_ct); obj = item.get(key_obj)
if ct is None or obj is None:
continue
out.append((int(ct), int(obj)))
return out
except Exception:
return []
def inspection_configuration_geojson(request, config_id):
from interventions.models import InspectionConfiguration
from django.db.models import Q
config = get_object_or_404(InspectionConfiguration, pk=config_id)
features = []
# Collect the ids of concerned categories (if any are configured)
concerned_category_ids = list(config.concerned_categories.values_list('pk', flat=True))
def _has_field(Model, field_name):
"""Check if a model has a specific DB field."""
return field_name in [f.name for f in Model._meta.get_fields()]
def _get_coords_and_geom(obj):
"""
Returns (lon, lat, geom_dict) for an object.
Priority: geojson field → lon/lat point → geom centroid
Returns None if no coordinates can be resolved.
"""
# Try the geojson text field first (pre-serialized geometry)
if hasattr(obj, 'geojson') and obj.geojson:
try:
geom_dict = json.loads(obj.geojson)
# Derive lon/lat from the geojson if not set directly
lon = obj.lon if obj.lon is not None else None
lat = obj.lat if obj.lat is not None else None
if lon is None and geom_dict.get('type') == 'Point':
lon, lat = geom_dict['coordinates'][0], geom_dict['coordinates'][1]
return lon, lat, geom_dict
except Exception:
pass
# Try lon/lat point
if obj.lon is not None and obj.lat is not None:
return obj.lon, obj.lat, {"type": "Point", "coordinates": [obj.lon, obj.lat]}
# Fallback: extract centroid from geom field
geom = getattr(obj, 'geom', None)
if geom is not None:
try:
centroid = geom.centroid.transform(4326, clone=True)
lon, lat = centroid.x, centroid.y
return lon, lat, {"type": "Point", "coordinates": [lon, lat]}
except Exception:
pass
return None, None, None
def _build_qs(Model):
"""Build queryset including objects with lon/lat OR a geom."""
has_geom = _has_field(Model, 'geom')
if has_geom:
return Model.objects.filter(
Q(lon__isnull=False, lat__isnull=False) | Q(geom__isnull=False)
)
return Model.objects.filter(lon__isnull=False, lat__isnull=False)
# 1) Add locations of chosen displayed_location_types
for ct in config.displayed_location_types.all():
Model = ct.model_class()
if Model:
qs = _build_qs(Model)
# Apply category filter only when categories are configured AND the model has the field
if concerned_category_ids and _has_field(Model, 'category'):
qs = qs.filter(category_id__in=concerned_category_ids)
for location_obj in qs:
lon, lat, geom_dict = _get_coords_and_geom(location_obj)
if geom_dict is None:
continue
features.append({
"type": "Feature",
"geometry": geom_dict,
"properties": {
"type": "location",
"ct": ct.id,
"id": location_obj.id,
"code": getattr(location_obj, 'code', ''),
"label": str(location_obj),
"name": location_obj.get_name() if hasattr(location_obj, 'get_name') else str(location_obj),
"lon": lon,
"lat": lat,
}
})
# 2) Add assets of chosen displayed_asset_types
for ct in config.displayed_asset_types.all():
Model = ct.model_class()
if Model:
qs = _build_qs(Model)
# Apply category filter when categories are configured AND the model has the field
if concerned_category_ids and _has_field(Model, 'category'):
qs = qs.filter(category_id__in=concerned_category_ids)
for asset_obj in qs:
lon, lat, geom_dict = _get_coords_and_geom(asset_obj)
if geom_dict is None:
continue
features.append({
"type": "Feature",
"geometry": geom_dict,
"properties": {
"type": "asset",
"ct": ct.id,
"id": asset_obj.id,
"code": getattr(asset_obj, 'code', ''),
"label": str(asset_obj),
"name": asset_obj.get_name() if hasattr(asset_obj, 'get_name') else (getattr(asset_obj, 'name_fr', '') or str(asset_obj)),
"lon": lon,
"lat": lat,
}
})
geojson_data = {
"type": "FeatureCollection",
"features": features
}
return JsonResponse(geojson_data)
def auto_check_if_eligible(user, intervention):
"""
Vérifie si l'utilisateur a le droit can_check_interventions sur le contrat de l'intervention.
Si oui, marque automatiquement l'intervention comme vérifiée (is_checked=True).
"""
if not intervention.contract or not getattr(intervention.contract, 'needs_checking', False):
return
try:
user_config = UserConfig.objects.filter(user=user).first()
if not user_config:
return
has_check_right = UserContractAccess.objects.filter(
user_config=user_config,
contract=intervention.contract,
can_check_interventions=True,
).exists()
if has_check_right and not intervention.is_checked:
intervention.is_checked = True
intervention.save(update_fields=['is_checked'])
except Exception:
pass
def _get_photos_required_since(intervention):
"""Retourne la date du premier passage au statut 'to_be_processed', ou None si aucun événement trouvé."""
event = InterventionTimeLine.objects.filter(
intervention=intervention,
to_status='to_be_processed',
).order_by('event_time').first()
return event.event_time if event else None
def _is_photo_required_for_user(user, intervention, user_config=None):
"""
Vérifie si l'obligation d'ajouter une photo s'applique à cet utilisateur pour cette intervention.
L'obligation s'applique uniquement si :
1. L'équipe assignée à l'intervention a photos_required=True.
2. L'utilisateur est un technicien (possède le rôle 'technician' et aucun rôle privilégié de gestion/admin).
"""
if not intervention:
return False
assigned_team = getattr(intervention, 'assigned_team', None)
if not assigned_team or not getattr(assigned_team, 'photos_required', False):
return False
if not user or not getattr(user, 'is_authenticated', False):
return False
if getattr(user, 'is_superuser', False):
return False
if user_config is None:
try:
user_config = UserConfig.objects.prefetch_related('roles').get(user=user)
except UserConfig.DoesNotExist:
return False
roles = _get_role_names(user_config)
privileged_roles = {'admin', 'top_manager', 'manager', 'controller', 'operator', 'external_manager'}
# L'obligation concerne uniquement les techniciens n'ayant pas de rôle privilégié (gestionnaire, admin...)
return 'technician' in roles and not bool(roles & privileged_roles)
def _get_user_short_name(user):
if not user:
return ""
first = getattr(user, "first_name", "") or ""
last = getattr(user, "last_name", "") or ""
initial = (last[0].upper() + ".") if last else ""
display = f"{first} {initial}".strip()
if not display:
return getattr(user, "username", "")
return display
def _get_available_occupation_members(user, intervention):
base_qs = CompanyMember.objects.select_related('user').prefetch_related('teams__company')
filters = Q()
if intervention.assigned_team_id:
filters |= Q(teams__id=intervention.assigned_team_id)
if intervention.assigned_provider_id:
filters |= Q(teams__company_id=intervention.assigned_provider_id)
if intervention.assigned_member_id:
filters |= Q(pk=intervention.assigned_member_id)
user_member = (CompanyMember.objects
.prefetch_related('teams__company')
.filter(user=user)
.first())
if user_member:
filters |= Q(teams__id__in=user_member.teams.values_list('id', flat=True))
filters |= Q(pk=user_member.pk)
if not filters:
return base_qs.none()
return base_qs.filter(filters).distinct().order_by('name')
def _serialize_occupation(occupation):
member = occupation.member
created_at_local = localtime(occupation.created_at) if occupation.created_at else None
updated_at_local = localtime(occupation.updated_at) if occupation.updated_at else None
return {
'id': occupation.id,
'member_id': occupation.member_id,
'member_name': getattr(member, 'name', ''),
'minutes': occupation.minutes,
'created_at': occupation.created_at.isoformat() if occupation.created_at else '',
'created_at_display': created_at_local.strftime('%d-%m-%Y %H:%M') if created_at_local else '',
'updated_at': occupation.updated_at.isoformat() if occupation.updated_at else '',
'updated_at_display': (updated_at_local.strftime('%d-%m-%Y %H:%M')
if updated_at_local and occupation.updated_at and occupation.updated_at != occupation.created_at
else ''),
'created_by_display': _get_user_short_name(occupation.created_by),
'updated_by_display': _get_user_short_name(occupation.updated_by),
}
def _serialize_note(note_instance, user):
author = note_instance.note_author
author_display = _get_user_short_name(author)
return {
'id': note_instance.id,
'note_author': getattr(author, 'username', ''),
'note_author_display': author_display,
'note_author_id': note_instance.note_author_id,
'note_time': note_instance.note_time,
'note_time_iso': note_instance.note_time.isoformat() if note_instance.note_time else '',
'content': str(linebreaksbr(note_instance.content)),
'content_raw': note_instance.content,
'content_html': str(linebreaksbr(note_instance.content)),
'is_owner': note_instance.note_author_id == getattr(user, 'id', None),
'updated_at': note_instance.updated_at,
'updated_by': _get_user_short_name(note_instance.updated_by) if note_instance.updated_by else '',
'note_type': note_instance.note_type,
'note_type_display': note_instance.get_note_type_display(),
'note_type_badge_class': note_instance.type_badge_class,
'note_type_icon_class': note_instance.type_icon_class,
'note_type_priority': note_instance.type_priority_value,
}
def _handle_notes_update(request, intervention, notes_data):
"""
Handle batch update of notes (new, updated, deleted).
Returns (success: bool, errors: list)
Note: Permission checks are done in the calling function (update_intervention)
"""
errors = []
success = False
try:
# Handle new notes
for note_data in notes_data.get('new_notes', []):
content = note_data.get('content', '').strip()
note_type = note_data.get('note_type', '')
if not content:
continue
try:
InterventionNote.objects.create(
intervention=intervention,
content=content,
note_type=note_type,
note_author=request.user,
note_time=timezone.now()
)
success = True
except Exception as e:
errors.append(_('Error creating note: %(error)s') % {'error': str(e)})
# Handle updated notes
for note_data in notes_data.get('updated_notes', []):
note_id = note_data.get('id')
content = note_data.get('content', '').strip()
note_type = note_data.get('note_type', '')
if not note_id or not content:
continue
try:
# Convert note_id to int if it's a string
note_id = int(note_id)
note = InterventionNote.objects.get(id=note_id, intervention=intervention)
# Check if user can edit this note
if not can_edit_note(request.user, request.user.config, note):
errors.append(_('You do not have permission to edit this note'))
continue
note.content = content
note.note_type = note_type
note.updated_by = request.user
note.updated_at = timezone.now()
note.save()
success = True
except InterventionNote.DoesNotExist:
errors.append(_('Note not found: %(note_id)s') % {'note_id': note_id})
except Exception as e:
errors.append(_('Error updating note: %(error)s') % {'error': str(e)})
# Handle deleted notes
for note_id in notes_data.get('deleted_notes', []):
try:
# Convert note_id to int if it's a string
note_id = int(note_id)
note = InterventionNote.objects.get(id=note_id, intervention=intervention)
# Check if user can delete this note
if note.note_author != request.user and not request.user.is_superuser:
errors.append(_('You do not have permission to delete this note'))
continue
note.delete()
success = True
except InterventionNote.DoesNotExist:
errors.append(_('Note not found: %(note_id)s') % {'note_id': note_id})
except Exception as e:
errors.append(_('Error deleting note: %(error)s') % {'error': str(e)})
except Exception as e:
errors.append(_('Error processing notes: %(error)s') % {'error': str(e)})
return success, errors
def _update_intervention_field(intervention, field, new_value, user_config=None):
"""
Helper function to update a single field on an intervention.
Returns (success: bool, error_message: str|None)
Args:
intervention: L'intervention à mettre à jour
field: Le nom du champ à mettre à jour
new_value: La nouvelle valeur
user_config: La configuration utilisateur (optionnel, utilisé pour vérifier les permissions)
"""
from common.hooks import hooks
success, error = hooks.run_inline_edit_validator(field, new_value, intervention)
if success is not None:
return success, error
if field == "title":
if len(new_value) <= 200:
intervention.title = new_value
return True, None
return False, _('Value for title must be limited to 200 characters')
elif field == "assigned_provider":
if not new_value:
intervention.assigned_provider = None
intervention.contract = None
return True, None
company = Company.objects.filter(id=new_value).first()
if company and intervention.assigned_provider != company:
# Vérifier l'accès aux contrats pour les utilisateurs limités aux contrats ou externes
if user_config and (user_config.limit_interventions_to_contracts or not user_config.is_intern):
has_access = UserContractAccess.objects.filter(
user_config=user_config,
contract__company=company,
can_view_interventions=True
).exists()
if not has_access:
return False, _('You do not have access to this provider')
# Remove the contract if the provider changes
intervention.contract = None
intervention.assigned_provider = company
return True, None
elif not company:
return False, _('Company not found')
return True, None # Same company, no change needed
elif field == "contract":
if not new_value:
intervention.contract = None
return True, None
contract = Contract.objects.filter(id=new_value).first()
if contract:
# Vérifier l'accès au contrat pour les utilisateurs limités aux contrats ou externes
if user_config and (user_config.limit_interventions_to_contracts or not user_config.is_intern):
has_access = UserContractAccess.objects.filter(
user_config=user_config,
contract=contract,
can_view_interventions=True
).exists()
if not has_access:
return False, _('You do not have access to this contract')
intervention.contract = contract
if contract.company and intervention.assigned_provider != contract.company:
intervention.assigned_provider = contract.company
return True, None
return False, _('Contract not found')
elif field == "maintain_type":
if new_value in dict(MAINTAIN_CHOICES).keys():
intervention.maintain_type = new_value
return True, None
return False, _('Maintain Type not found')
elif field == "type":
allowed_types = dict(get_type_choices(intervention.thematic)).keys()
if new_value in allowed_types:
intervention.type = new_value
return True, None
return False, _('Type not found')
elif field == "origin_type":
if new_value in dict(ORIGIN_TYPE_CHOICES).keys():
intervention.origin_type = new_value
return True, None
return False, _('Origin Type not found')
elif field == "project":
if new_value == "":
# Supprimer l'association projet-intervention si elle existe
ProjectIntervention.objects.filter(intervention=intervention).delete()
return True, None
else:
try:
project_intervention = ProjectIntervention.objects.get_or_create(project_id=new_value, intervention=intervention)[0]
return True, None
except Exception:
return False, _('Project not found')
elif field == "osiris":
if new_value.isdigit():
intervention.osiris = int(new_value)
return True, None
return False, _('Value for Osiris must be a positive integer')
elif field == "pmo_number":
value_str = str(new_value).strip() if new_value is not None else ""
if value_str == "":
intervention.pmo_number = None
return True, None
if value_str.isdigit():
intervention.pmo_number = int(value_str)
return True, None
return False, _('Value for PMO must be a positive integer')
elif field == "linked_folder":
if len(new_value) <= 1024:
intervention.linked_folder = new_value
return True, None
return False, _('Value for folder must be limited to 1024 characters')
elif field == "member_assignment":
if new_value.startswith("team-"):
team_id = int(new_value.split("-")[1])
assigned_team = CompanyTeam.objects.filter(id=team_id).first()
if not assigned_team:
return False, _("Selected team not found")
# Vérifier que l'équipe est associée au prestataire assigné
if intervention.assigned_provider and assigned_team.company != intervention.assigned_provider:
return False, _("Selected team does not belong to the assigned provider")
# Vérifier que l'équipe est associée à la thématique de l'intervention
if intervention.thematic and not assigned_team.thematics.filter(id=intervention.thematic.id).exists():
return False, _("Selected team is not associated with the thematic of this intervention")
intervention.assigned_member = None
intervention.assigned_team = assigned_team
return True, None
elif new_value.startswith("member-"):
member_id = int(new_value.split("-")[1])
assigned_member = CompanyMember.objects.filter(id=member_id).first()
if not assigned_member:
return False, _("Selected member not found")
# Vérifier que le membre appartient à une équipe du prestataire assigné
matching_teams = assigned_member.teams.all()
if intervention.assigned_provider:
matching_teams = matching_teams.filter(company=intervention.assigned_provider)
if not matching_teams.exists():
return False, _("Selected member does not belong to the assigned provider")
# Vérifier que l'équipe du membre est associée à la thématique de l'intervention
if intervention.thematic:
thematic_teams = matching_teams.filter(thematics=intervention.thematic)
if not thematic_teams.exists():
return False, _("Selected member's team is not associated with the thematic of this intervention")
assigned_team = thematic_teams.first()
else:
assigned_team = matching_teams.first()
intervention.assigned_team = assigned_team
intervention.assigned_member = assigned_member
return True, None
return False, _("Invalid assignment value.")
elif field == "category":
normalized_value = str(new_value).strip() if new_value is not None else ""
if not normalized_value:
intervention.asset_category = None
return True, None
if not normalized_value.isdigit():
return False, _('Value for category must be a positive integer')
category_id = int(normalized_value)
categories_qs = AssetCategory.objects.all()
allowed_category_ids = _get_allowed_asset_category_ids(intervention)
if allowed_category_ids:
categories_qs = categories_qs.filter(id__in=allowed_category_ids)
category = categories_qs.filter(id=category_id).first()
if not category:
return False, _('Selected category is not allowed for this intervention')
intervention.asset_category = category
return True, None
elif field == "symptom":
if new_value == "":
intervention.symptom = None
return True, None
if new_value.isdigit():
symptom = Symptom.objects.filter(id=int(new_value)).first()
if symptom:
intervention.symptom = symptom
return True, None
return False, _('Symptom not found')
return False, _('Value for Symptom must be a positive integer')
elif field == "priority":
# Get priority choices from the model field
priority_choices_dict = dict(Intervention._meta.get_field('priority').choices)
if new_value in priority_choices_dict.keys():
intervention.priority = new_value
return True, None
return False, _('Priority not found')
elif field == "source_category":
if new_value == "":
intervention.source_category = None
return True, None
if new_value.isdigit():
source_category = SourceCategory.objects.filter(id=int(new_value)).first()
if source_category:
intervention.source_category = source_category
return True, None
return False, _('Source category not found')
return False, _('Value for Source category must be a positive integer')
elif field == "source_ref":
if len(new_value) <= 255:
intervention.source_ref = new_value if new_value else None
return True, None
return False, _('Value for source_ref must be limited to 255 characters')
elif field == "source_detail":
if len(new_value) <= 255:
intervention.source_detail = new_value if new_value else None
return True, None
return False, _('Value for source_detail must be limited to 255 characters')
elif field == "description":
if len(new_value) <= 1024:
intervention.description = new_value
return True, None
return False, _('Value for description must be limited to 1024 characters')
elif field == "order":
if new_value == "":
intervention.order = None
return True, None
if new_value.isdigit():
order = ContractOrder.objects.filter(id=int(new_value)).first()
if order:
intervention.order = order
return True, None
return False, _('Order not found')
return False, _('Value for Order must be a positive integer')
elif field == "processed_via_other_intervention":
if isinstance(new_value, bool):
intervention.processed_via_other_intervention = new_value
return True, None
# Handle string values from form POST
if new_value in ['true', 'True', '1', 'on']:
intervention.processed_via_other_intervention = True
return True, None
elif new_value in ['false', 'False', '0', 'off', '']:
intervention.processed_via_other_intervention = False
return True, None
return False, _('Value for processed_via_other_intervention must be boolean')
elif field == "address":
intervention.address = new_value if new_value else None
return True, None
elif field == "lat":
try:
intervention.lat = float(new_value) if new_value else None
if intervention.lat and intervention.lon:
pt = Point(intervention.lon, intervention.lat, srid=4326)
pt.transform(3812)
intervention.geom = to_multipolygon(pt.buffer(1))
return True, None
except (ValueError, TypeError):
return False, _('Value for lat must be a decimal number')
elif field == "lon":
try:
intervention.lon = float(new_value) if new_value else None
if intervention.lat and intervention.lon:
pt = Point(intervention.lon, intervention.lat, srid=4326)
pt.transform(3812)
intervention.geom = to_multipolygon(pt.buffer(1))
return True, None
except (ValueError, TypeError):
return False, _('Value for lon must be a decimal number')
# Note: planification and realization are NOT handled by this helper as they
# require request.POST access. They remain in the main update_intervention view.
return False, _('Unknown field: %(field)s') % {'field': field}
def _serialize_timeline_event(event):
localized_time = localtime(event.event_time)
return {
"event_id": event.id,
"event_time_formatted": localized_time.strftime("%d/%m/%Y %H:%M"),
"event_time_iso": localized_time.strftime("%Y-%m-%dT%H:%M"),
"event_description": event.event_description,
"event_description_truncated": Truncator(event.event_description).chars(100),
"event_type": event.event_type,
"event_type_label": event.get_event_type_display(),
"event_is_future": event.event_time >= now(),
}
def _apply_planification_side_effects(intervention, event):
mapping = {
'planned_begin': 'planned_begin_time',
'planned_end': 'planned_end_time',
'expected_begin': 'expected_begin_time',
'expected_end': 'expected_end_time',
}
target_attr = mapping.get(event.event_type)
if target_attr:
setattr(intervention, target_attr, event.event_time)
return target_attr
return None
def _clear_planification_side_effects(intervention, event_type, event_time):
mapping = {
'planned_begin': 'planned_begin_time',
'planned_end': 'planned_end_time',
'expected_begin': 'expected_begin_time',
'expected_end': 'expected_end_time',
}
target_attr = mapping.get(event_type)
if target_attr and getattr(intervention, target_attr) == event_time:
setattr(intervention, target_attr, None)
return target_attr
return None
def _apply_realization_side_effects(intervention, event):
mapping = {
'real_begin': 'begin_time',
'real_end': 'end_time',
}
target_attr = mapping.get(event.event_type)
if target_attr:
setattr(intervention, target_attr, event.event_time)
return target_attr
return None
def _clear_realization_side_effects(intervention, event_type, event_time):
mapping = {
'real_begin': 'begin_time',
'real_end': 'end_time',
}
target_attr = mapping.get(event_type)
if target_attr and getattr(intervention, target_attr) == event_time:
setattr(intervention, target_attr, None)
return target_attr
return None
# Legacy handlers for complex fields (planification, realization) that need request.POST
def _handle_planification_update(request, intervention):
"""Handle planification timeline updates"""
action = request.POST.get("action")
from django.utils.timezone import now as tz_now
current_time = tz_now()
# Delete an event
if action == "delete":
event_id = request.POST.get("event_id")
if not event_id:
return JsonResponse({"success": False, "error": _("Event ID is required.")}, status=400)
try:
event = InterventionPlanificationTimeLine.objects.get(id=event_id, intervention=intervention)
except InterventionPlanificationTimeLine.DoesNotExist:
return JsonResponse({"success": False, "error": _("Event not found.")}, status=404)
if event.event_time < current_time:
return JsonResponse({"success": False, "error": _("Seuls les événements futurs peuvent être modifiés ou supprimés.")}, status=400)
cleared_attr = _clear_planification_side_effects(intervention, event.event_type, event.event_time)
event.delete()
if cleared_attr:
intervention.save(update_fields=[cleared_attr])
return JsonResponse({"success": True})
if action == "update":
event_id = request.POST.get("event_id")
if not event_id:
return JsonResponse({"success": False, "error": _("Event ID is required.")}, status=400)
try:
event = InterventionPlanificationTimeLine.objects.get(id=event_id, intervention=intervention)
except InterventionPlanificationTimeLine.DoesNotExist:
return JsonResponse({"success": False, "error": _("Event not found.")}, status=404)
if event.event_time < current_time:
return JsonResponse({"success": False, "error": _("Seuls les événements futurs peuvent être modifiés ou supprimés.")}, status=400)
event_time_str = request.POST.get("event_time")
event_type = request.POST.get("event_type")
event_description = (request.POST.get("event_description") or "").strip()
if not event_time_str or not event_description or not event_type:
return JsonResponse({"success": False, "error": _("Champs requis manquants.")}, status=400)
event_time = parse_datetime(event_time_str)
if event_time is None:
return JsonResponse({"success": False, "error": _("Format datetime invalide")}, status=400)
if is_naive(event_time):
event_time = make_aware(event_time, get_current_timezone())
previous_type = event.event_type
previous_time = event.event_time
event.event_user = request.user
event.event_time = event_time
event.event_type = event_type
event.event_description = event_description
event.save()
cleared_attr = _clear_planification_side_effects(intervention, previous_type, previous_time)
applied_attr = _apply_planification_side_effects(intervention, event)
if cleared_attr or applied_attr:
update_fields = {attr for attr in [cleared_attr, applied_attr] if attr}
intervention.save(update_fields=list(update_fields))
payload = _serialize_timeline_event(event)
return JsonResponse({"success": True, **payload})
# Add a new event
event_time_str = request.POST.get("event_time")
event_type = request.POST.get("event_type")
event_description = request.POST.get("event_description")
if not event_time_str or not event_description or not event_type:
return JsonResponse({"success": False, "error": _("Champs requis manquants.")}, status=400)
try:
event_time = parse_datetime(event_time_str)
if event_time is None:
raise ValueError(_("Format datetime invalide"))
if is_naive(event_time):
event_time = make_aware(event_time, get_current_timezone())
event = InterventionPlanificationTimeLine.objects.create(
intervention=intervention,
event_user=request.user,
event_time=event_time,
event_type=event_type,
event_description=event_description
)
updated_attr = _apply_planification_side_effects(intervention, event)
if updated_attr:
intervention.save(update_fields=[updated_attr])
payload = _serialize_timeline_event(event)
return JsonResponse({"success": True, **payload})
except ValueError as exc:
return JsonResponse({"success": False, "error": str(exc)}, status=400)
except Exception as e:
return JsonResponse({"success": False, "error": str(e)}, status=500)
def _handle_realization_update(request, intervention):
"""Handle realization timeline updates"""
action = request.POST.get("action")
# Delete an event
if action == "delete":
event_id = request.POST.get("event_id")
if not event_id:
return JsonResponse({"success": False, "error": _("Event ID is required.")}, status=400)
try:
event = InterventionRealizationTimeLine.objects.get(id=event_id, intervention=intervention)
except InterventionRealizationTimeLine.DoesNotExist:
return JsonResponse({"success": False, "error": _("Event not found.")}, status=404)
cleared_attr = _clear_realization_side_effects(intervention, event.event_type, event.event_time)
event.delete()
if cleared_attr:
intervention.save(update_fields=[cleared_attr])
return JsonResponse({"success": True})
if action == "update":
event_id = request.POST.get("event_id")
if not event_id:
return JsonResponse({"success": False, "error": _("Event ID is required.")}, status=400)
try:
event = InterventionRealizationTimeLine.objects.get(id=event_id, intervention=intervention)
except InterventionRealizationTimeLine.DoesNotExist:
return JsonResponse({"success": False, "error": _("Event not found.")}, status=404)
event_time_str = request.POST.get("event_time")
event_description = request.POST.get("event_description")
event_type = request.POST.get("event_type")
if not event_time_str or not event_description or not event_type:
return JsonResponse({"success": False, "error": _("Champs requis manquants.")}, status=400)
event_time = parse_datetime(event_time_str)
if event_time is None:
return JsonResponse({"success": False, "error": _("Format datetime invalide")}, status=400)
if is_naive(event_time):
event_time = make_aware(event_time, get_current_timezone())
previous_type = event.event_type
previous_time = event.event_time
event.event_user = request.user
event.event_time = event_time
event.event_type = event_type
event.event_description = event_description
event.save()
cleared_attr = _clear_realization_side_effects(intervention, previous_type, previous_time)
applied_attr = _apply_realization_side_effects(intervention, event)
if cleared_attr or applied_attr:
update_fields = {attr for attr in [cleared_attr, applied_attr] if attr}
intervention.save(update_fields=list(update_fields))
payload = _serialize_timeline_event(event)
return JsonResponse({"success": True, **payload})
# Add a new event
event_time_str = request.POST.get("event_time")
event_description = request.POST.get("event_description")
event_type = request.POST.get("event_type")
if not event_time_str or not event_description or not event_type:
return JsonResponse({"success": False, "error": _("Champs requis manquants.")}, status=400)
try:
event_time = parse_datetime(event_time_str)
if event_time is None:
raise ValueError(_("Format datetime invalide"))
if is_naive(event_time):
event_time = make_aware(event_time, get_current_timezone())
event = InterventionRealizationTimeLine.objects.create(
intervention=intervention,
event_user=request.user,
event_time=event_time,
event_type=event_type,
event_description=event_description
)
updated_attr = _apply_realization_side_effects(intervention, event)
if updated_attr:
intervention.save(update_fields=[updated_attr])
payload = _serialize_timeline_event(event)
return JsonResponse({"success": True, **payload})
except ValueError as exc:
return JsonResponse({"success": False, "error": str(exc)}, status=400)
except Exception as e:
return JsonResponse({"success": False, "error": str(e)}, status=500)
def resolve_selected_assets(draft_selected_assets):
"""
draft_selected_assets: [{'model': 'trafficlightpole', 'object_id': 1}, ...]
-> [{'id', 'asset_type', 'code', 'name', 'lon', 'lat'}, ...]
"""
if not draft_selected_assets:
return []
# 1) Grouper par model, mémoriser l'ordre pour le rendu
groups = defaultdict(list)
order = []
for item in draft_selected_assets:
model = str(item.get('model', '')).lower().strip()
oid = int(item.get('object_id'))
groups[model].append(oid)
order.append((model, oid))
# 2) Résoudre les ContentTypes d'un coup
model_names = list(groups.keys())
cts = {ct.model: ct for ct in ContentType.objects.filter(model__in=model_names)}
# 3) Charger en masse et indexer
obj_maps = {}
for model, ids in groups.items():
ct = cts.get(model)
if not ct:
continue
Model = ct.model_class()
# set(ids) pour éviter les doublons
objs = Model.objects.filter(id__in=set(ids))
obj_maps[model] = {o.id: o for o in objs}
# 4) Recomposer dans l'ordre demandé
out = []
for model, oid in order:
obj = obj_maps.get(model, {}).get(oid)
if not obj:
continue
out.append({
"id": obj.id,
"asset_type": model, # ex. 'trafficlightpole'
"code": getattr(obj, "code", ""),
"name": obj.get_name() if hasattr(obj, "get_name") else getattr(obj, "name", ""),
"lon": getattr(obj, "lon", None), # AbstractGeoAsset → lon/lat si point
"lat": getattr(obj, "lat", None),
})
return out
def batch_fetch_assets_with_locations(selected_assets_array):
"""
selected_assets_array: [{'id': int, 'asset_type': 'trafficlightpole', ...}, ...]
-> retourne:
{
'trafficlightpole': { 12: <TrafficLightPole ...>, ... },
'trafficlightcable': { 34: <TrafficLightCable ...>, ... },
...
}
avec les FKs de localisation prefetchées si disponibles (select_related).
"""
by_model = defaultdict(set)
for a in selected_assets_array or []:
by_model[str(a['asset_type']).lower()].add(int(a['id']))
models_map = {}
for model_name, ids in by_model.items():
ct = ContentType.objects.filter(app_label='assets', model=model_name.lower()).first() or ContentType.objects.filter(model=model_name.lower()).first()
if not ct:
continue
Model = ct.model_class()
if not Model:
continue
# détecter quelles FKs de localisation existent réellement sur le modèle
related_fields = []
for cand in CANDIDATE_LOCATION_FKS:
try:
f = Model._meta.get_field(cand)
if isinstance(f, ForeignKey):
related_fields.append(cand)
except Exception:
pass
qs = Model.objects.filter(id__in=ids)
if related_fields:
qs = qs.select_related(*related_fields)
models_map[model_name] = {obj.id: obj for obj in qs}
return models_map
def to_multipolygon(geom, srid=None):
if not geom or geom.empty:
return None
# Harmonise le SRID si besoin
if srid is not None and getattr(geom, "srid", None) != srid:
geom.srid = srid
gtype = geom.geom_type # 'Polygon', 'MultiPolygon', 'GeometryCollection', etc.
if gtype == 'Polygon':
return MultiPolygon(geom)
if gtype == 'MultiPolygon':
return geom
if gtype == 'GeometryCollection':
polys = [g for g in geom if g.geom_type in ['Polygon', 'MultiPolygon']]
if polys:
all_polys = []
for p in polys:
if p.geom_type == 'MultiPolygon':
all_polys.extend(list(p))
else:
all_polys.append(p)
return MultiPolygon(all_polys)
return None
if gtype in ['Point', 'MultiPoint', 'LineString', 'MultiLineString']:
buffered = geom.buffer(1)
if buffered.geom_type == 'Polygon':
return MultiPolygon(buffered)
elif buffered.geom_type == 'MultiPolygon':
return buffered
raise ValueError(f"Type géométrique non supporté: {gtype}")
def resolve_coords_from_hierarchy(loc, max_depth=10):
"""
Remonte la hiérarchie parent_location pour trouver lon/lat/geom.
Retourne (lon, lat, geom) - les valeurs peuvent être None.
"""
current = loc
depth = 0
lon, lat, geom = None, None, None
while current is not None and depth < max_depth:
current_lon = getattr(current, "lon", None) or getattr(current, "longitude", None)
current_lat = getattr(current, "lat", None) or getattr(current, "latitude", None)
current_geom = getattr(current, "geom", None)
if lon is None and current_lon is not None:
lon = current_lon
if lat is None and current_lat is not None:
lat = current_lat
if geom is None and current_geom is not None:
geom = current_geom
if lon is not None and lat is not None and geom is not None:
break
parent = getattr(current, "parent_location", None)
if parent is None:
struct = getattr(current, "structure", None)
parent = getattr(struct, "main_location", None) if struct else None
current = parent
depth += 1
return lon, lat, geom
def resolve_intervention_contract_and_provider(thematic, symptom, asset=None, location=None, user=None):
"""
Détermine le contrat et le prestataire associés pour une intervention, ainsi que le flag force_to_be_approved
si l'utilisateur externe n'a pas accès au contrat résolu.
Retourne un tuple (contract, assigned_provider, force_to_be_approved)
"""
contract = None
assigned_provider = None
force_to_be_approved = False
thematic_code = thematic.code if hasattr(thematic, 'code') else (str(thematic) if thematic else None)
# 1. Routage direct (thématique 'structures') - prioritaire sur les contrats par défaut
if not contract and thematic_code == 'structures':
structure = None
if asset:
if isinstance(asset, Structure):
structure = asset
elif hasattr(asset, 'structure') and asset.structure:
structure = asset.structure
if not structure and location:
if isinstance(location, Structure):
structure = location
elif hasattr(location, 'structure') and location.structure:
structure = location.structure
if structure and getattr(structure, 'code', None):
routing = DirectContractRouting.objects.filter(
thematic='structures',
asset_code=structure.code,
symptom=symptom
).select_related('contract__company').first()
if not routing:
routing = DirectContractRouting.objects.filter(
thematic='structures',
asset_code=structure.code,
symptom__isnull=True
).select_related('contract__company').first()
if routing and routing.contract:
contract = routing.contract
if not assigned_provider:
assigned_provider = contract.company
# 2. Contrat actif lié à la localisation/équipement et au maintenance_type
if not contract:
mt = symptom.maintenance_type if symptom else None
if thematic_code == 'trafficlights' and mt:
intersection_id = None
if location:
if isinstance(location, TrafficLightIntersection):
intersection_id = location.id
elif hasattr(location, 'intersection_id') and location.intersection_id:
intersection_id = location.intersection_id
if not intersection_id and asset:
if isinstance(asset, TrafficLightIntersection):
intersection_id = asset.id
elif hasattr(asset, 'intersection_id') and asset.intersection_id:
intersection_id = asset.intersection_id
if intersection_id:
tl_contract = TrafficLightContract.objects.filter(
intersection_id=intersection_id,
maintenance_type=mt,
status='active'
).select_related('contract__company').first()
if tl_contract and tl_contract.contract:
contract = tl_contract.contract
if not assigned_provider:
assigned_provider = contract.company
elif thematic_code == 'structures':
structure_id = None
if location:
if isinstance(location, Structure):
structure_id = location.id
elif hasattr(location, 'structure_id') and location.structure_id:
structure_id = location.structure_id
if not structure_id and asset:
if isinstance(asset, Structure):
structure_id = asset.id
elif hasattr(asset, 'structure_id') and asset.structure_id:
structure_id = asset.structure_id
if structure_id:
struct_contract = None
if mt:
struct_contract = StructureContract.objects.filter(
structure_id=structure_id,
maintenance_type=mt,
status='active'
).select_related('contract__company').first()
if not struct_contract:
struct_contracts = StructureContract.objects.filter(
structure_id=structure_id,
status='active'
).select_related('contract__company')
if struct_contracts.count() == 1:
struct_contract = struct_contracts.first()
if struct_contract and struct_contract.contract:
contract = struct_contract.contract
if not assigned_provider:
assigned_provider = contract.company
elif thematic_code == 'clean' and mt:
clean_loc_id = None
if location:
if isinstance(location, CleanLocation):
clean_loc_id = location.id
elif hasattr(location, 'location_id') and location.location_id:
clean_loc_id = location.location_id
if not clean_loc_id and asset:
if isinstance(asset, CleanLocation):
clean_loc_id = asset.id
elif hasattr(asset, 'location_id') and asset.location_id:
clean_loc_id = asset.location_id
if clean_loc_id:
clean_contract = CleanLocationContract.objects.filter(
location_id=clean_loc_id,
maintenance_type=mt,
status='active'
).select_related('contract__company').first()
if clean_contract and clean_contract.contract:
contract = clean_contract.contract
if not assigned_provider:
assigned_provider = contract.company
elif thematic_code == 'its' and mt:
its_contract = None
if asset and isinstance(asset, ITSGeoAsset):
its_contract = ITSGeoAssetContract.objects.filter(
asset_id=asset.id,
maintenance_type=mt,
status='active'
).select_related('contract__company').first()
elif location and isinstance(location, ITSLocation):
its_contract = ITSLocationContract.objects.filter(
location_id=location.id,
maintenance_type=mt,
status='active'
).select_related('contract__company').first()
if its_contract and its_contract.contract:
contract = its_contract.contract
if not assigned_provider:
assigned_provider = contract.company
# 3. Contrat / prestataire direct du symptôme (fallback par défaut du symptôme)
if not contract and symptom:
if symptom.contract:
contract = symptom.contract
assigned_provider = symptom.provider or (contract.company if contract else None)
elif symptom.provider:
assigned_provider = symptom.provider
# 4. Fallback contrat par défaut de l'utilisateur (limit_interventions_to_contracts)
user_config = None
if user:
try:
user_config = getattr(user, 'config', None) or UserConfig.objects.filter(user=user).first()
except Exception:
user_config = None
if not contract and user_config and user_config.limit_interventions_to_contracts:
default_contract = user_config.get_default_contract()
if default_contract:
contract = default_contract
if not assigned_provider:
assigned_provider = default_contract.company
# 5. Vérification d'accès contrat pour utilisateur externe (limit_interventions_to_contracts)
if user_config and not user_config.is_intern and user_config.limit_interventions_to_contracts:
if symptom:
if symptom.contract_id:
has_access = UserContractAccess.objects.filter(
user_config=user_config,
contract_id=symptom.contract_id,
).exists()
if not has_access:
force_to_be_approved = True
elif symptom.maintenance_type:
if contract:
has_access = UserContractAccess.objects.filter(
user_config=user_config,
contract_id=contract.id,
).exists()
if not has_access:
force_to_be_approved = True
else:
force_to_be_approved = True
return contract, assigned_provider, force_to_be_approved
def resolve_asset_spatial_data(asset=None, location=None):
"""
Extrait et calcule la géométrie (MultiPolygon SRID 3812), les coordonnées (lon, lat WGS84 4326),
l'adresse, le location_code et l'objet de localisation parent.
Retourne un dictionnaire:
{
'geom': MultiPolygon | None (SRID 3812),
'lon': float | None (WGS84 4326),
'lat': float | None (WGS84 4326),
'address': str | None,
'location_code': str | None,
'location_obj': Model | None,
}
"""
geom = None
lon = None
lat = None
address = None
location_code = None
loc_obj = location
# 1. Déterminer l'objet localisation parent si non fourni
if not loc_obj and asset:
loc_obj = (
asset.get_location() if hasattr(asset, 'get_location') else None
) or getattr(asset, 'street', None) or getattr(asset, 'location', None) or getattr(asset, 'intersection', None) or getattr(asset, 'structure', None) or getattr(asset, 'section', None)
# 2. Extraction adresse et location_code
if asset:
address = getattr(asset, 'address', None)
if not address and hasattr(asset, 'get_address'):
try:
address = asset.get_address()
except Exception:
address = None
location_code = getattr(asset, 'code', None)
if not address and loc_obj:
address = getattr(loc_obj, 'address', None)
if not address and hasattr(loc_obj, 'get_name'):
try:
address = loc_obj.get_name()
except Exception:
address = None
if loc_obj and hasattr(loc_obj, 'code') and loc_obj.code:
location_code = loc_obj.code
# 3. Extraction de la géométrie source
raw_geom = getattr(asset, 'geom', None) if asset else None
if raw_geom is None and loc_obj:
raw_geom = getattr(loc_obj, 'geom', None)
if raw_geom is None:
_, _, raw_geom = resolve_coords_from_hierarchy(loc_obj)
# Coordonnées directes (lat/lon)
if asset:
lon = getattr(asset, 'lon', None) or getattr(asset, 'longitude', None)
lat = getattr(asset, 'lat', None) or getattr(asset, 'latitude', None)
if lon is None and lat is None and hasattr(asset, 'get_coordinates'):
try:
coords = asset.get_coordinates()
if coords and len(coords) >= 2:
lat, lon = coords[0], coords[1]
except Exception:
pass
if (lon is None or lat is None) and loc_obj:
h_lon, h_lat, _ = resolve_coords_from_hierarchy(loc_obj)
if lon is None:
lon = h_lon
if lat is None:
lat = h_lat
# 4. Traitement de la géométrie vers MultiPolygon SRID 3812
if raw_geom is not None and not raw_geom.empty:
try:
geom_3812 = raw_geom.clone()
if getattr(geom_3812, 'srid', None) and geom_3812.srid != 3812:
geom_3812.transform(3812)
elif not getattr(geom_3812, 'srid', None):
geom_3812.srid = 3812
gtype = geom_3812.geom_type
if gtype in ['Point', 'MultiPoint', 'LineString', 'MultiLineString']:
buffered = geom_3812.buffer(1)
geom = to_multipolygon(buffered, srid=3812)
elif gtype == 'Polygon':
geom = MultiPolygon(geom_3812)
geom.srid = 3812
elif gtype == 'MultiPolygon':
geom = geom_3812
elif gtype == 'GeometryCollection':
polys = [g for g in geom_3812 if g.geom_type in ['Polygon', 'MultiPolygon']]
if polys:
all_polys = []
for p in polys:
if p.geom_type == 'MultiPolygon':
all_polys.extend(list(p))
else:
all_polys.append(p)
geom = MultiPolygon(all_polys)
geom.srid = 3812
# Extraire lon/lat si non encore définis
if lon is None or lat is None:
geom_4326 = raw_geom.clone()
if getattr(geom_4326, 'srid', None) != 4326:
if not getattr(geom_4326, 'srid', None):
geom_4326.srid = 3812
geom_4326.transform(4326)
centroid = geom_4326.centroid
if lon is None:
lon = centroid.x
if lat is None:
lat = centroid.y
except Exception as e:
logger.warning(f"Impossible de convertir la géométrie : {e}")
# Si pas de géométrie valide mais lon/lat sont présents : créer un buffer de 1m
if geom is None and lon is not None and lat is not None:
try:
pt = Point(float(lon), float(lat), srid=4326)
pt.transform(3812)
geom = MultiPolygon(pt.buffer(1))
geom.srid = 3812
except Exception as e:
logger.warning(f"Impossible de générer le buffer géométrique depuis lon/lat : {e}")
return {
'geom': geom,
'lon': float(lon) if lon is not None else None,
'lat': float(lat) if lat is not None else None,
'address': address,
'location_code': location_code,
'location_obj': loc_obj,
}
def resolve_asset_category(asset, symptom=None):
"""
Résout la catégorie d'asset appropriée pour une intervention liée à un asset.
"""
category = getattr(asset, 'category', None)
if not category:
# Vérifier si l'asset a un modèle avec catégorie
for model_attr in ('pole_model', 'lantern_model', 'controller_model', 'cabinet_model', 'detector_model', 'radar_model', 'model'):
model_obj = getattr(asset, model_attr, None)
if model_obj and getattr(model_obj, 'category', None):
category = model_obj.category
break
# Si toujours pas de catégorie et qu'un symptôme est fourni avec une seule catégorie
if not category and symptom:
symptom_cats = list(symptom.asset_categories.all())
if len(symptom_cats) == 1:
category = symptom_cats[0]
return category
def is_user_same_as_provider(user, intervention) -> bool:
"""
Vérifie si l'utilisateur appartient à la même organisation que le prestataire assigné à l'intervention.
"""
if not user or not getattr(user, 'is_authenticated', False):
return False
if getattr(user, 'is_superuser', False):
return False
user_config = getattr(user, 'config', None)
# Les utilisateurs internes (ex: SPRB / régie) ne sont pas considérés comme le prestataire externe
if user_config and getattr(user_config, 'is_intern', False):
return False
# Récupérer l'ID des sociétés prestataires liées à l'intervention
provider_ids = set()
if getattr(intervention, 'assigned_provider_id', None):
provider_ids.add(intervention.assigned_provider_id)
if getattr(intervention, 'contract_id', None) and getattr(intervention.contract, 'company_id', None):
provider_ids.add(intervention.contract.company_id)
if getattr(intervention, 'assigned_team_id', None) and getattr(intervention.assigned_team, 'company_id', None):
provider_ids.add(intervention.assigned_team.company_id)
if not provider_ids:
return False
# 1. Vérifier via user_config.company
if user_config and user_config.company_id in provider_ids:
return True
# 2. Vérifier via CompanyMember (équipes de la société)
from contracts.models import CompanyMember
if CompanyMember.objects.filter(user=user, teams__company_id__in=provider_ids).exists():
return True
# 3. Vérifier via UserContractAccess
if user_config and user_config.contracts.filter(company_id__in=provider_ids).exists():
return True
return False
def get_contract_recipients(contract, intervention=None) -> List[str]:
"""Retourne une liste unique d'adresses email associées au contrat."""
# En dev/test, redirige tous les emails vers l'adresse de test
email_override = getattr(settings, 'EMAIL_OVERRIDE', None)
if email_override:
return [email_override]
if not contract:
return []
if hasattr(contract, 'send_email_notifications') and not contract.send_email_notifications:
return []
if getattr(contract, 'email_only_if_urgent', False) and intervention and getattr(intervention, 'priority', None) != '1':
return []
emails = set()
# 1) Champ texte sur le contrat (séparé par ; ou ,)
if hasattr(contract, 'notification_emails') and contract.notification_emails:
for e in re.split(r'[;,]', contract.notification_emails):
e = e.strip()
if e:
emails.add(e)
# 2) Relations de contacts (ex: contract.contacts)
if hasattr(contract, 'contacts'):
qs = contract.contacts.all()
# Filtre facultatif si vous avez un flag pour recevoir les rapports
if hasattr(qs.model, 'receive_intervention_reports'):
qs = qs.filter(receive_intervention_reports=True)
emails.update([c.email for c in qs if getattr(c, 'email', None)])
return [e for e in emails if e]
def get_contract_cc_recipients(contract, intervention=None) -> List[str]:
"""Retourne une liste unique d'adresses email en copie (CC) associées au contrat."""
# En dev/test, ne pas utiliser de CC pour éviter les envois externes
email_override = getattr(settings, 'EMAIL_OVERRIDE', None)
if email_override:
return []
if not contract:
return []
if hasattr(contract, 'send_email_notifications') and not contract.send_email_notifications:
return []
if getattr(contract, 'email_only_if_urgent', False) and intervention and getattr(intervention, 'priority', None) != '1':
return []
emails = set()
# Champ texte CC sur le contrat (séparé par ; ou ,)
if hasattr(contract, 'cc_notification_emails') and contract.cc_notification_emails:
for e in re.split(r'[;,]', contract.cc_notification_emails):
e = e.strip()
if e:
emails.add(e)
# Toujours ajouter l'email par défaut en CC
emails.add(getattr(settings, 'EMAIL_TECH_SUPPORT', 'bm.perm.tech@sprb.brussels'))
return [e for e in emails if e]
def get_intervention_attachments(
intervention,
max_total_bytes: int = 15 * 1024 * 1024, # ~15 Mo (ajustez au besoin)
exclude_filenames: Iterable[str] = (),
) -> List[Tuple[str, bytes, str]]:
"""
Retourne une liste de tuples (filename, content_bytes, mimetype) pour les fichiers
déjà joints à l'intervention, en respectant un plafond global de taille.
exclude_filenames: liste de noms de fichiers à ne pas inclure (ex. le PDF qu'on vient de générer).
"""
attachments: List[Tuple[str, bytes, str]] = []
total = 0
seen = set(exclude_filenames)
# Conventions possibles côté modèle
candidate_related_names = ("documents",) #"attachments", "files", "related_documents")
# Récupération des objets "document"
candidates = []
for rel in candidate_related_names:
if hasattr(intervention, rel):
qs = getattr(intervention, rel).all()
# Si votre modèle a un flag is_active
if hasattr(qs.model, "is_active"):
qs = qs.filter(is_active=True)
candidates.extend(list(qs))
# Détecter le FileField sur chaque objet document
for doc in candidates:
file_obj = None
for attr in ("file", "document", "content"):
if hasattr(doc, attr):
file_obj = getattr(doc, attr)
break
if not file_obj or not getattr(file_obj, "name", None):
continue
filename = os.path.basename(file_obj.name)
if not filename or filename in seen:
continue
try:
data = file_obj.read() # lit depuis le storage
except Exception:
continue
if not data:
continue
# Respect du plafond global
if total + len(data) > max_total_bytes:
# on ignore si dépassement ; vous pouvez plutôt "break" ou garder une file d'attente
continue
mimetype, _ = mimetypes.guess_type(filename)
attachments.append((filename, data, mimetype or "application/octet-stream"))
total += len(data)
seen.add(filename)
return attachments