loko/loko/mobile/views.py

1695 lines
70 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import re
from functools import wraps
from django.shortcuts import render, redirect
from django.contrib import messages
from django.contrib.auth.decorators import login_not_required, login_required
from django.http import JsonResponse, HttpResponse
from django.conf import settings
from django.contrib.staticfiles import finders as staticfiles_finders
from django.views.decorators.cache import never_cache
from django.utils.translation import get_language, gettext as _
from pathlib import Path
from common.models import UserConfig, Thematic, UserThematics
from common.utils import check_thematic_access
from assets.models import AssetCategory, NatureAssetModel
from assets.models.nature import BULB_TYPE_CHOICES
def stock_access_required(view_func):
"""
Decorator to ensure the user has one of the allowed stock roles:
admin, stock_manager, manager, controller, top_manager.
"""
@wraps(view_func)
def _wrapped_view(request, *args, **kwargs):
if not request.user.is_authenticated:
return redirect('%s?next=%s' % (settings.LOGIN_URL, request.path))
try:
user_config = request.user.config
roles = set(user_config.roles.values_list('name', flat=True))
allowed_roles = {'admin', 'stock_manager', 'manager', 'controller', 'top_manager'}
if request.user.is_superuser or roles.intersection(allowed_roles):
return view_func(request, *args, **kwargs)
except (UserConfig.DoesNotExist, AttributeError):
pass
return render(request, "403.html", status=403)
return _wrapped_view
def index(request):
"""Render the mobile landing page."""
user_config = None
try:
user_config = UserConfig.objects.get(user=request.user)
roles = list(user_config.roles.values_list('name', flat=True))
# Les techniciens (et observateurs) sont redirigés directement sur leur page de travail
# lors de leur arrivée initiale dans la session (/mobile/), mais peuvent accéder à la page d'accueil ensuite
# (ex: clic sur l'icône Accueil ?home=1, navigation interne depuis la session, etc.)
wants_home = request.GET.get('home') == '1' or request.session.get('mobile_home_visited', False)
if not wants_home:
# Vérifie si l'utilisateur n'a qu'un seul rôle et que c'est 'technician'
if len(roles) == 1 and roles[0] == 'technician':
request.session['mobile_home_visited'] = True
return redirect('/mobile/interventions/')
# Vérifie si l'utilisateur n'a qu'un seul rôle et que c'est 'observer'
if len(roles) == 1 and roles[0] == 'observer':
request.session['mobile_home_visited'] = True
return redirect('/mobile/observations/')
else:
request.session['mobile_home_visited'] = True
except UserConfig.DoesNotExist:
# If no user config exists, return unauthorized (403)
return render(request, "403.html", status=403)
# Check if user has permission to edit green surfaces (user config thematics)
can_edit_green_surfaces = False
try:
nature_thematic = Thematic.objects.get(code='nature')
can_edit_green_surfaces = user_config.is_intern and UserThematics.objects.filter(
user_config=user_config,
thematic=nature_thematic,
can_edit_assets=True
).exists()
except Thematic.DoesNotExist:
pass
# Check if user has access to at least one warehouse (view permission)
from stock.permissions import get_user_accessible_warehouses
has_stock_role = request.user.is_superuser or any(r in roles for r in ['admin', 'stock_manager', 'manager', 'controller', 'top_manager'])
can_access_stock = has_stock_role and get_user_accessible_warehouses(request.user).exists()
# Check if user can add repair interventions for structures thematic (technicians)
can_add_structures_repair = False
default_thematic = user_config.default_thematic
if default_thematic and default_thematic.code == 'structures':
is_technician = 'technician' in roles
if is_technician:
can_add_structures_repair = UserThematics.objects.filter(
user_config=user_config,
thematic=default_thematic,
can_edit_interventions=True
).exists()
# Check if user (internal admin/manager/controller) can add interventions across all thematics
can_add_intervention_mobile = user_config.is_intern and any(r in roles for r in ['admin', 'manager', 'controller'])
# Check if user is an inspector and get inspection configurations
is_inspector = 'inspector' in roles
inspection_configs = []
if is_inspector:
from interventions.models import InspectionConfiguration
from interventions.permissions import can_user_create_inspection_for_contract
configs = list(
InspectionConfiguration.objects.select_related('thematic')
.prefetch_related('contracts_providers__contract', 'operation_templates')
)
is_admin = 'admin' in roles
for config in configs:
associated_contracts = [cp.contract for cp in config.contracts_providers.all()]
if not associated_contracts:
inspection_configs.append(config)
elif is_admin:
inspection_configs.append(config)
else:
for contract in associated_contracts:
if can_user_create_inspection_for_contract(request.user, contract):
inspection_configs.append(config)
break
can_inspect = user_config.get_inspectable_thematics().exists()
can_view_inspections = can_inspect or user_config.has_role('viewer')
# Check if user has permission to view assets across any thematic
can_view_assets = request.user.is_superuser or UserThematics.objects.filter(
user_config=user_config,
can_view_assets=True
).exists()
return render(request, "mobile/mobile_index.html", {
'can_edit_green_surfaces': can_edit_green_surfaces,
'can_access_stock': can_access_stock,
'can_add_structures_repair': can_add_structures_repair,
'can_add_intervention_mobile': can_add_intervention_mobile,
'is_inspector': is_inspector,
'inspection_configs': inspection_configs,
'can_inspect': can_inspect,
'can_view_inspections': can_view_inspections,
'can_view_assets': can_view_assets,
})
@login_not_required
def install(request):
"""Render installation instructions for the mobile app."""
return render(request, "mobile/mobile_install.html")
@never_cache
def mobile_controls_service_worker(request):
"""Sert le service worker des contrôles depuis /mobile/controls/service-worker.js.
Ce chemin donne au SW un scope naturel sur /mobile/controls/ qui couvre
toutes les pages mobiles de détail et de liste des contrôles.
"""
rel = 'controls/service-worker.js'
# 1. Finders (développement et production avec STATICFILES_DIRS)
abs_path = staticfiles_finders.find(rel)
if abs_path:
try:
content = Path(abs_path).read_text(encoding='utf-8')
return HttpResponse(content, content_type='application/javascript; charset=utf-8')
except (FileNotFoundError, IOError):
pass
# 2. STATIC_ROOT (après collectstatic en production)
try:
content = (Path(settings.STATIC_ROOT) / rel).read_text(encoding='utf-8')
return HttpResponse(content, content_type='application/javascript; charset=utf-8')
except (FileNotFoundError, IOError, TypeError):
pass
# 3. Chemin source app (fallback ultime)
try:
content = (settings.BASE_DIR / 'controls' / 'static' / rel).read_text(encoding='utf-8')
except (FileNotFoundError, IOError):
content = 'console.error("Controls service worker introuvable");'
return HttpResponse(content, content_type='application/javascript; charset=utf-8')
@check_thematic_access('nature')
def green_surfaces_edit_mobile(request):
"""Mobile interface for editing green surfaces (nature assets)."""
# Get categories and models for nature thematic
categories = list(AssetCategory.objects.filter(thematic__code='nature'))
categories.sort(key=lambda m: m.get_name().lower())
models = list(NatureAssetModel.objects.filter(category__code='NAT_GREEN_AREAS'))
models.sort(key=lambda m: (m.get_name() or "").lower())
context = {
'categories': categories,
'models': models,
'bulb_type_choices': sorted(BULB_TYPE_CHOICES, key=lambda x: x[1].lower()),
}
return render(request, "mobile/green_surfaces_edit_v2.html", context)
@check_thematic_access('nature')
def green_surfaces_edit_custom_draw(request):
"""Mobile interface for editing green surfaces with custom draw module (test page)."""
# Get categories and models for nature thematic
categories = list(AssetCategory.objects.filter(thematic__code='nature'))
categories.sort(key=lambda m: m.get_name().lower())
models = list(NatureAssetModel.objects.filter(category__code='NAT_GREEN_AREAS'))
models.sort(key=lambda m: (m.get_name() or "").lower())
context = {
'categories': categories,
'models': models,
}
return render(request, "mobile/green_surfaces_edit_custom_draw.html", context)
@login_required
@stock_access_required
def stock_index_mobile(request):
"""Mobile landing page for the stock module."""
from stock.models import PreparationOrder, ProductionOrder
from stock.permissions import get_user_accessible_warehouses
preparation_orders_count = PreparationOrder.objects.filter(
status__in=['to_process', 'pending']
).count()
accessible_warehouses = get_user_accessible_warehouses(request.user)
production_orders_count = ProductionOrder.objects.filter(
warehouse__id__in=accessible_warehouses,
status__in=['to_process', 'in_production'],
).count()
return render(request, "stock/mobile/stock_index_mobile.html", {
'preparation_orders_count': preparation_orders_count,
'production_orders_count': production_orders_count,
})
@login_required
@stock_access_required
def stock_products_mobile(request):
"""Mobile list of stock products."""
from stock.models import Product, ProductCategory
from django.db.models import Q
q = request.GET.get('q', '').strip()
category_id = request.GET.get('category', '').strip()
products = Product.objects.select_related('category').order_by('name')
if q:
products = products.filter(Q(name__icontains=q) | Q(code__icontains=q))
if category_id:
products = products.filter(category_id=category_id)
categories = ProductCategory.objects.order_by('name_fr')
return render(request, "stock/mobile/stock_products_mobile.html", {
'products': products[:200],
'q': q,
'category_id': category_id,
'categories': categories,
})
@login_required
@stock_access_required
def qr_scanner(request):
"""Page du scanner QR Code mobile."""
return render(request, 'mobile/qr_scanner.html')
@login_required
@stock_access_required
def qr_resolve(request):
"""
Résout le contenu d'un QR code scanné et retourne l'URL de redirection en JSON.
Format attendu: LOKO:<type>:<identifiant> ou STREETUP:<type>:<identifiant> (rétro-compatibilité)
Types supportés : product, order
"""
code = request.GET.get('code', '').strip()
if not (code.startswith('LOKO:') or code.startswith('STREETUP:')):
return JsonResponse({'error': _('QR Code non reconnu.')}, status=400)
parts = code.split(':', 2)
if len(parts) != 3:
return JsonResponse({'error': _('Format de QR Code invalide.')}, status=400)
_, qr_type, identifier = parts
if qr_type == 'product':
from stock.models import Product
try:
product = Product.objects.get(code=identifier)
return JsonResponse({
'url': f'/mobile/stock/products/{product.pk}/',
'label': f"▣ {product.name} ({product.code})",
})
except Product.DoesNotExist:
return JsonResponse({'error': _(f'Produit « {identifier} » introuvable.')}, status=404)
if qr_type == 'order':
from stock.models import PurchaseOrder
try:
order = PurchaseOrder.objects.get(uuid=identifier)
return JsonResponse({
'url': f'/stock/purchase-orders/{order.pk}/',
'label': f'📦 {order.code or "#" + str(order.pk)}',
})
except (PurchaseOrder.DoesNotExist, Exception):
return JsonResponse({'error': _(f'Commande introuvable.')}, status=404)
if qr_type == 'serializeditem':
from stock.models import SerializedItem
try:
si = SerializedItem.objects.select_related('product').get(serial_number=identifier)
return JsonResponse({
'url': f'/mobile/stock/serialized/{si.pk}/',
'label': f'🔖 {si.product.name} — {si.serial_number}',
})
except SerializedItem.DoesNotExist:
return JsonResponse({'error': _(f'Article tracé « {identifier} » introuvable.')}, status=404)
if qr_type == 'stocklocation':
from stock.models import WarehouseLocation
try:
loc = WarehouseLocation.objects.select_related('warehouse').get(code=identifier)
return JsonResponse({
'url': f'/stock/warehouse-locations/{loc.pk}/',
'label': f'📍 {loc.warehouse.name} — {loc.name}',
})
except WarehouseLocation.DoesNotExist:
return JsonResponse({'error': _(f'Emplacement « {identifier} » introuvable.')}, status=404)
if qr_type == 'preparation':
from stock.models import PreparationOrder
try:
order = PreparationOrder.objects.get(order_number=identifier)
return JsonResponse({
'url': f'/mobile/stock/preparation-orders/{order.pk}/',
'label': f'📋 Bon de préparation {order.order_number}',
})
except PreparationOrder.DoesNotExist:
return JsonResponse({'error': _(f'Bon de préparation « {identifier} » introuvable.')}, status=404)
return JsonResponse({'error': _(f'Type de QR Code inconnu : {qr_type}.')}, status=400)
@login_required
@stock_access_required
def stock_product_detail_mobile(request, product_id):
"""Mobile detail page for a stock product."""
from stock.models import Product, Stock, PurchaseOrderItem, StockMovement
from common.models import UserConfig
from django.shortcuts import get_object_or_404
try:
user_config = request.user.config
is_stock_manager = user_config.has_role('stock_manager') or user_config.has_role('admin')
except (UserConfig.DoesNotExist, AttributeError):
is_stock_manager = False
product = get_object_or_404(Product.objects.select_related('category'), id=product_id)
stocks = list(Stock.objects.filter(product=product).select_related(
'location__warehouse'
).order_by('location__warehouse__name', 'location__name'))
from stock.permissions import user_can_administer_warehouse
for stock in stocks:
stock.can_administer = user_can_administer_warehouse(request.user, stock.location.warehouse)
pending_orders = PurchaseOrderItem.objects.filter(
product=product,
purchase_order__status__in=['pending', 'ordered']
).select_related('purchase_order__supplier', 'purchase_order__warehouse', 'location')
recent_movements = StockMovement.objects.filter(
product=product
).select_related(
'source_location__warehouse', 'destination_location__warehouse',
'batch', 'batch__preparation_order', 'batch__purchase_order',
).order_by('-date')[:20]
order_history = PurchaseOrderItem.objects.filter(
product=product,
purchase_order__status__in=['received', 'cancelled']
).select_related(
'purchase_order__supplier', 'purchase_order__warehouse'
).order_by('-purchase_order__date_ordered')[:30]
serialized_items = None
if product.is_serialized:
from stock.models import SerializedItem
serialized_items = SerializedItem.objects.filter(product=product).select_related(
'current_holder', 'current_location__warehouse'
).order_by('serial_number')
return render(request, 'stock/mobile/product_detail_mobile.html', {
'product': product,
'stocks': stocks,
'pending_orders': pending_orders,
'recent_movements': recent_movements,
'order_history': order_history,
'is_stock_manager': is_stock_manager,
'serialized_items': serialized_items,
})
@login_required
@stock_access_required
def stock_serialized_item_detail_mobile(request, pk):
"""Mobile detail page for a serialized (tracked) item."""
from stock.models import SerializedItem, WarehouseLocation
from stock.permissions import user_can_manage_products
from contracts.models import CompanyMember
from django.shortcuts import get_object_or_404
item = get_object_or_404(
SerializedItem.objects.select_related(
'product__category', 'current_location__warehouse', 'current_holder', 'created_by'
).prefetch_related('current_holder__teams'),
pk=pk,
)
events = item.events.select_related(
'person', 'from_location__warehouse', 'to_location__warehouse', 'created_by'
)
locations = WarehouseLocation.objects.select_related('warehouse').order_by('warehouse__name', 'name')
members = CompanyMember.objects.prefetch_related('teams').order_by('name')
return render(request, 'stock/mobile/serialized_item_detail_mobile.html', {
'item': item,
'events': events,
'locations': locations,
'members': members,
'can_manage': user_can_manage_products(request.user),
})
@login_required
@stock_access_required
def stock_preparation_orders_mobile(request):
from stock.models import PreparationOrder
from django.db import models
selected_status = request.GET.get('status', '').strip()
selected_priority = request.GET.get('priority', '').strip()
search_q = request.GET.get('q', '').strip()
preparation_orders = PreparationOrder.objects.select_related('intervention').all()
if selected_status:
preparation_orders = preparation_orders.filter(status=selected_status)
if selected_priority:
preparation_orders = preparation_orders.filter(priority=selected_priority)
if search_q:
q_filter = (
models.Q(order_number__icontains=search_q) |
models.Q(intervention__code__icontains=search_q) |
models.Q(intervention__title__icontains=search_q) |
models.Q(requester__icontains=search_q)
)
if search_q.isdigit():
q_filter |= models.Q(intervention__id=int(search_q))
preparation_orders = preparation_orders.filter(q_filter)
preparation_orders = preparation_orders.prefetch_related(
'preparationorderitem_set__product'
).order_by('-date')[:200]
return render(request, "stock/mobile/stock_preparation_orders_mobile.html", {
'preparation_orders': preparation_orders,
'selected_status': selected_status,
'selected_priority': selected_priority,
'search_q': search_q,
'preparation_order_status_choices': [
('to_process', _("A traiter")),
('pending', _("En attente")),
('processed', _("Traité")),
('cancelled', _("Annulé")),
],
'preparation_order_priority_choices': [
('1', _("Urgent")),
('2', _("Elevé")),
('3', _("Normal")),
('4', _("Bas")),
],
})
@login_required
@stock_access_required
def stock_production_orders_mobile(request):
"""Mobile list of production orders (bons de confection)."""
from stock.models import ProductionOrder
from stock.permissions import get_user_accessible_warehouses
accessible_warehouses = get_user_accessible_warehouses(request.user)
production_orders = ProductionOrder.objects.filter(
warehouse__id__in=accessible_warehouses
).select_related('warehouse').order_by('-created_at')[:200]
return render(request, "stock/mobile/stock_production_orders_mobile.html", {
'production_orders': production_orders,
})
@login_required
@stock_access_required
def stock_production_order_detail_mobile(request, pk):
"""Mobile detail page for a production order (bon de confection)."""
from django.shortcuts import get_object_or_404
from stock.models import ProductionOrder
from stock.permissions import get_user_accessible_warehouses
from common.models import UserConfig
accessible_warehouses = get_user_accessible_warehouses(request.user)
order = get_object_or_404(
ProductionOrder.objects.select_related(
'warehouse', 'preparation_order', 'output_product', 'output_location',
).prefetch_related('documents').filter(warehouse__id__in=accessible_warehouses),
pk=pk,
)
try:
user_config = request.user.config
can_manage = user_config.has_role('stock_manager') or user_config.has_role('admin')
except UserConfig.DoesNotExist:
can_manage = False
components_status = order.get_components_status()
can_start = order.can_start()
# Extraire la description libre (depuis le FK ou depuis les notes pré-remplies)
other_item_description = None
if order.preparation_order_other_item_id:
other_item_description = order.preparation_order_other_item.description
else:
notes = order.notes or ''
if 'Issu de' in notes and ': ' in notes:
other_item_description = notes.split(': ', 1)[1]
return render(request, "stock/mobile/stock_production_order_detail_mobile.html", {
'order': order,
'components_status': components_status,
'can_start': can_start,
'can_manage': can_manage,
'other_item_description': other_item_description,
})
@login_required
@stock_access_required
def stock_movements_mobile(request):
"""Mobile list of stock movements (entrées / sorties)."""
from stock.models import StockMovementBatch, StockMovement
from stock.permissions import get_user_accessible_warehouses
from django.db.models import Q, Prefetch
accessible_warehouses = get_user_accessible_warehouses(request.user)
batches = StockMovementBatch.objects.filter(
Q(movements__source_location__warehouse__id__in=accessible_warehouses) |
Q(movements__destination_location__warehouse__id__in=accessible_warehouses)
).distinct().select_related('person', 'company', 'created_by').prefetch_related(
Prefetch(
'movements',
queryset=StockMovement.objects.select_related(
'product', 'source_location', 'destination_location'
)
)
).order_by('-date')[:100]
return render(request, "stock/mobile/stock_movements_mobile.html", {
'batches': batches,
})
@login_required
@stock_access_required
def new_stock_entry_mobile(request):
"""Crée une nouvelle entrée de stock depuis l'interface mobile."""
from stock.views import _new_stock_batch_view
return _new_stock_batch_view(
request, 'in',
template_name="stock/mobile/new_stock_batch_mobile.html",
cancel_url_name="mobile:stock_movements_mobile",
success_redirect_fn=lambda req, batch: redirect(
'mobile:stock_movement_batch_detail_mobile', batch_id=batch.pk
),
)
@login_required
@stock_access_required
def new_stock_exit_mobile(request):
"""Crée une nouvelle sortie de stock depuis l'interface mobile."""
from stock.views import _new_stock_batch_view
return _new_stock_batch_view(
request, 'out',
template_name="stock/mobile/new_stock_batch_mobile.html",
cancel_url_name="mobile:stock_movements_mobile",
success_redirect_fn=lambda req, batch: redirect(
'mobile:stock_movement_batch_detail_mobile', batch_id=batch.pk
),
)
@login_required
@stock_access_required
def new_stock_transfer_mobile(request):
"""Crée un nouveau transfert de stock depuis l'interface mobile."""
from stock.views import _new_stock_batch_view
return _new_stock_batch_view(
request, 'transfer',
template_name="stock/mobile/new_stock_batch_mobile.html",
cancel_url_name="mobile:stock_movements_mobile",
success_redirect_fn=lambda req, batch: redirect(
'mobile:stock_movement_batch_detail_mobile', batch_id=batch.pk
),
)
@login_required
@stock_access_required
def production_order_create_mobile(request):
"""Crée un nouveau bon de confection depuis l'interface mobile."""
from stock.models import ProductionOrder, PreparationOrderItem, PreparationOrderOtherItem
from stock.forms import ProductionOrderForm, ProductionOrderComponentFormSet
from common.models import UserConfig
try:
user_config = request.user.config
is_stock_manager = user_config.has_role('stock_manager') or user_config.has_role('admin')
except UserConfig.DoesNotExist:
is_stock_manager = False
if not is_stock_manager:
messages.error(request, _("Vous n'avez pas la permission d'accéder à cette page."))
return redirect('mobile:stock_production_orders_mobile')
initial = {}
other_item_context = None
prep_item_context = None
prep_id = request.GET.get('preparation_order')
other_item_id = request.GET.get('other_item')
prep_item_id = request.GET.get('prep_item')
if prep_id:
initial['preparation_order'] = prep_id
try:
default_warehouse = request.user.config.get_default_warehouse()
if default_warehouse:
initial['warehouse'] = default_warehouse
except Exception:
pass
if prep_item_id:
try:
prep_item = PreparationOrderItem.objects.select_related(
'preparation_order', 'product__category'
).get(pk=prep_item_id)
prep_item_context = prep_item
if not prep_id:
initial['preparation_order'] = prep_item.preparation_order_id
initial['quantity'] = prep_item.quantity
initial['notes'] = _(
"Issu de l'article {product} du bon de préparation #{order_number}"
).format(
product=prep_item.product.name,
order_number=prep_item.preparation_order.order_number,
)
except PreparationOrderItem.DoesNotExist:
pass
if other_item_id:
try:
other_item = PreparationOrderOtherItem.objects.select_related(
'preparation_order', 'sign_panel_type', 'category'
).get(pk=other_item_id)
other_item_context = other_item
if not prep_id:
initial['preparation_order'] = other_item.preparation_order_id
initial['quantity'] = other_item.quantity
initial['output_product_name'] = other_item.description[:300]
initial['notes'] = _(
"Issu de l'item #{id} du bon de préparation #{order_number}: {desc}"
).format(
id=other_item.id,
order_number=other_item.preparation_order.order_number,
desc=other_item.description,
)
except PreparationOrderOtherItem.DoesNotExist:
pass
if request.method == 'POST':
form = ProductionOrderForm(request.POST, request.FILES)
formset = ProductionOrderComponentFormSet(request.POST, prefix='components')
if form.is_valid() and formset.is_valid():
order = form.save(commit=False)
if prep_item_context:
order.preparation_order_item = prep_item_context
if other_item_context:
order.preparation_order_other_item = other_item_context
order.save()
form.save_m2m()
formset.instance = order
formset.save()
from stock.models import ProductionOrderDocument
for f in request.FILES.getlist('new_documents'):
ProductionOrderDocument.objects.create(
production_order=order,
file=f,
name=f.name,
)
messages.success(request, _("Bon de confection créé avec succès."))
return redirect('mobile:stock_production_order_detail_mobile', pk=order.pk)
else:
form = ProductionOrderForm(initial=initial)
formset = ProductionOrderComponentFormSet(prefix='components')
return render(request, 'stock/mobile/production_order_form_mobile.html', {
'form': form,
'formset': formset,
'action': 'create',
'other_item_context': other_item_context,
'prep_item_context': prep_item_context,
})
@login_required
@stock_access_required
def stock_movement_batch_detail_mobile(request, batch_id):
"""Mobile detail page for a stock movement batch."""
from stock.models import StockMovementBatch, StockBatchSignature
from stock.permissions import get_user_accessible_warehouses
from django.shortcuts import get_object_or_404
from django.db.models import Q
import base64
from django.core.files.base import ContentFile
batch = get_object_or_404(
StockMovementBatch.objects.select_related(
'person',
'company',
'contract', 'contract__company',
'created_by',
'preparation_order',
'purchase_order', 'purchase_order__supplier',
).prefetch_related(
'person__teams',
'movements__product',
'movements__source_location__warehouse',
'movements__destination_location__warehouse',
'signatures__signed_by',
'documents',
),
pk=batch_id,
)
# Vérifier si l'utilisateur peut gérer au moins un entrepôt impliqué dans ce lot
manageable_warehouses = get_user_accessible_warehouses(request.user, can_manage=True)
batch_warehouse_ids = set(
batch.movements.filter(source_location__warehouse__isnull=False)
.values_list('source_location__warehouse_id', flat=True)
) | set(
batch.movements.filter(destination_location__warehouse__isnull=False)
.values_list('destination_location__warehouse_id', flat=True)
)
can_add_signature = bool(
manageable_warehouses.filter(id__in=batch_warehouse_ids).exists()
)
if request.method == 'POST':
if not can_add_signature:
messages.error(request, _("Vous n'avez pas la permission d'ajouter une signature à ce lot."))
return redirect('mobile:stock_movement_batch_detail_mobile', batch_id=batch.pk)
signature_data = request.POST.get('signature_data', '').strip()
signature_name = request.POST.get('signature_name', '').strip() or None
image_file = None
if signature_data and signature_data.startswith('data:image/'):
header, encoded = signature_data.split(',', 1)
image_bytes = base64.b64decode(encoded)
image_file = ContentFile(image_bytes, name=f'signature_batch_{batch.pk}_{batch.signatures.count() + 1}.png')
elif request.FILES.get('signature'):
image_file = request.FILES['signature']
if image_file:
sig = StockBatchSignature(batch=batch, signed_by=request.user, signature_name=signature_name)
sig.image.save(image_file.name if hasattr(image_file, 'name') else f'sig_{batch.pk}.png', image_file, save=True)
messages.success(request, _('Signature enregistrée avec succès.'))
return redirect('mobile:stock_movement_batch_detail_mobile', batch_id=batch.pk)
return render(request, 'stock/mobile/stock_movement_batch_detail_mobile.html', {
'batch': batch,
'can_add_signature': can_add_signature,
})
@login_required
@stock_access_required
def stock_inventories_mobile(request):
"""Mobile list of inventories."""
from stock.models import Inventory
from stock.permissions import get_user_accessible_warehouses
accessible_warehouses = get_user_accessible_warehouses(request.user)
inventories = Inventory.objects.filter(
warehouse_location__warehouse__id__in=accessible_warehouses
).select_related('warehouse_location__warehouse').order_by('-created_at')[:200]
return render(request, "stock/mobile/stock_inventories_mobile.html", {
'inventories': inventories,
})
@login_required
@stock_access_required
def stock_inventory_detail_mobile(request, pk):
"""Mobile detail page for an inventory (saisie des quantités comptées)."""
from django.shortcuts import get_object_or_404
from stock.models import Inventory
from stock.permissions import get_user_accessible_warehouses, user_can_validate_inventory
accessible_warehouses = get_user_accessible_warehouses(request.user)
inventory = get_object_or_404(
Inventory.objects.filter(
warehouse_location__warehouse__id__in=accessible_warehouses
).select_related("warehouse_location__warehouse"),
pk=pk,
)
items = inventory.items.select_related("product", "validated_by").order_by("product__name")
warehouse = inventory.warehouse_location.warehouse
can_validate = user_can_validate_inventory(request.user, warehouse)
from stock.permissions import user_can_manage_warehouse
can_change_status = user_can_manage_warehouse(request.user, warehouse)
return render(request, "stock/mobile/stock_inventory_detail_mobile.html", {
"inventory": inventory,
"items": items,
"can_validate": can_validate,
"can_change_status": can_change_status,
})
@login_required
@stock_access_required
def stock_new_inventory_mobile(request):
"""Mobile form to create a new inventory."""
from stock.models import Inventory, WarehouseLocation, Warehouse
from stock.permissions import get_user_accessible_warehouses
accessible_warehouses = get_user_accessible_warehouses(request.user)
locations = (
WarehouseLocation.objects.filter(warehouse__id__in=accessible_warehouses)
.select_related('warehouse')
.order_by('warehouse__name', 'name')
)
# Group locations by warehouse for the template
warehouses_with_locations = {}
for loc in locations:
wh = loc.warehouse
warehouses_with_locations.setdefault(wh, []).append(loc)
grouped_locations = [
{'warehouse': wh, 'locations': locs}
for wh, locs in warehouses_with_locations.items()
]
error = None
performed_by = ''
location_id = ''
if request.method == 'POST':
location_id = request.POST.get('warehouse_location', '').strip()
performed_by = request.POST.get('performed_by', '').strip()
if not location_id:
error = _("Veuillez sélectionner un emplacement.")
else:
location = WarehouseLocation.objects.filter(
pk=location_id,
warehouse__id__in=accessible_warehouses,
).first()
if not location:
error = _("Emplacement invalide ou non accessible.")
else:
inventory = Inventory.objects.create(
warehouse_location=location,
performed_by=performed_by,
status=Inventory.STATUS_IN_PROGRESS,
)
# Pré-remplir les articles avec les produits existants à cet emplacement
from stock.models import Stock, InventoryItem
existing_stocks = (
Stock.objects.filter(location=location, quantity__gt=0)
.select_related('product')
.order_by('product__name')
)
for s in existing_stocks:
InventoryItem.objects.create(
inventory=inventory,
product=s.product,
expected_quantity=s.quantity,
counted_quantity=s.quantity,
)
messages.success(request, _("Inventaire créé avec succès."))
return redirect('mobile:stock_inventory_detail_mobile', pk=inventory.pk)
return render(request, "stock/mobile/stock_new_inventory_mobile.html", {
'grouped_locations': grouped_locations,
'error': error,
'performed_by': performed_by,
'location_id': location_id,
})
@login_required
@stock_access_required
def stock_purchases_mobile(request):
"""Mobile list of purchase orders (commandes)."""
from stock.models import PurchaseOrder
from stock.permissions import get_user_accessible_warehouses
accessible_warehouses = get_user_accessible_warehouses(request.user)
purchase_orders = PurchaseOrder.objects.filter(
warehouse__id__in=accessible_warehouses
).select_related('supplier', 'warehouse').order_by('-date_ordered')[:200]
return render(request, "stock/mobile/stock_purchases_mobile.html", {
'purchase_orders': purchase_orders,
})
@login_required
@stock_access_required
def stock_purchase_order_detail_mobile(request, pk):
"""Mobile detail page for a purchase order (bon de commande)."""
from django.shortcuts import get_object_or_404
from stock.models import PurchaseOrder, PurchaseOrderItem
from stock.permissions import get_user_accessible_warehouses
from common.models import UserConfig
from django.db.models import Sum as _Sum
accessible_warehouses = get_user_accessible_warehouses(request.user)
purchase_order = get_object_or_404(
PurchaseOrder.objects.filter(
warehouse__id__in=accessible_warehouses
).select_related('supplier', 'warehouse')
.prefetch_related('items__product', 'items__location__warehouse'),
pk=pk,
)
qr_code = purchase_order.generate_qr_code()
items_with_stock = []
for item in purchase_order.items.all():
items_with_stock.append({
'item': item,
})
try:
user_config = request.user.config
can_manage_purchase = user_config.can_manage_purchase_orders_any()
except UserConfig.DoesNotExist:
can_manage_purchase = False
return render(request, "stock/mobile/stock_purchase_order_detail_mobile.html", {
'purchase_order': purchase_order,
'qr_code': qr_code,
'items_with_stock': items_with_stock,
'can_manage_purchase': can_manage_purchase,
})
@login_required
@stock_access_required
def stock_preparation_order_detail_mobile(request, pk):
"""Mobile detail page for a preparation order (bon de préparation)."""
from django.shortcuts import get_object_or_404
from stock.models import PreparationOrder, Stock
from stock.permissions import get_user_accessible_warehouses
from common.models import UserConfig
preparation_order = get_object_or_404(
PreparationOrder.objects.select_related(
'intervention',
'intervention__intervention_manager',
).prefetch_related(
'preparationorderitem_set__product__category',
'preparationorderitem_set__production_orders',
'other_items__category',
'other_items__sign_panel_type',
'other_items__production_orders',
'documents',
),
pk=pk,
)
# Ajout de documents via POST
if request.method == 'POST' and request.FILES.getlist('documents'):
from stock.models import PreparationOrderDocument
for f in request.FILES.getlist('documents'):
PreparationOrderDocument.objects.create(
preparation_order=preparation_order,
file=f,
name=f.name,
)
messages.success(request, _('Document(s) ajouté(s) avec succès.'))
return redirect('mobile:stock_preparation_order_detail_mobile', pk=pk)
from stock.models import PreparationOrderItem, WarehouseLocation
executed_summary = preparation_order.get_executed_output_summary()
executed_by_loc = preparation_order.get_executed_outputs_by_product_and_location()
executed_totals = {p_id: info['net'] for p_id, info in executed_summary.items()}
has_executed_outputs = any(info['out'] > 0 for info in executed_summary.values())
is_reopened = preparation_order.status == 'to_process' and has_executed_outputs
# S'assurer qu'une ligne PreparationOrderItem avec quantity=0 existe pour tout produit sorti mais supprimé du bon
for p_id, loc_dict in executed_by_loc.items():
executed_qty = sum(loc_dict.values())
if executed_qty > 0:
PreparationOrderItem.objects.get_or_create(
preparation_order=preparation_order,
product_id=p_id,
defaults={'quantity': 0}
)
items_with_stock = []
for item in preparation_order.preparationorderitem_set.filter(quantity__gt=0):
product = item.product
available_stock = product.total_stock
reserved_stock = product.reserved_quantity
prod_order = item.production_orders.first()
prod_summary = executed_summary.get(product.id, {'out': 0, 'in': 0, 'net': 0})
executed_qty = prod_summary['net']
gross_out = prod_summary['out']
gross_in = prod_summary['in']
delta = item.quantity - executed_qty
items_with_stock.append({
'item': item,
'available_stock': available_stock,
'reserved_stock': reserved_stock,
'is_sufficient': available_stock >= item.quantity,
'production_order': prod_order,
'executed_qty': executed_qty,
'gross_out': gross_out,
'gross_in': gross_in,
'delta': delta,
'locations': Stock.objects.filter(product=product, quantity__gt=0).select_related('location__warehouse'),
})
deleted_items_to_return = []
for item in preparation_order.preparationorderitem_set.filter(quantity=0):
executed_qty = executed_totals.get(item.product_id, 0)
if executed_qty > 0:
product = item.product
loc_dict = executed_by_loc.get(product.id, {})
loc_names = []
for l_id, q in loc_dict.items():
loc_obj = WarehouseLocation.objects.filter(pk=l_id).select_related('warehouse').first()
if loc_obj:
loc_names.append(f"{loc_obj.name} ({loc_obj.warehouse.name}): {q}")
deleted_items_to_return.append({
'item': item,
'product': product,
'executed_qty': executed_qty,
'location_info': ", ".join(loc_names),
})
try:
user_config = request.user.config
can_manage = user_config.has_role('stock_manager') or user_config.has_role('admin') or user_config.can_manage_preparation_orders_any()
can_cancel = user_config.can_manage_preparation_orders_any()
can_manage_purchase = user_config.can_manage_purchase_orders_any()
can_manage_stock = (
request.user.is_superuser
or user_config.has_role('admin')
or user_config.has_role('stock_manager')
or user_config.get_accessible_warehouses(can_manage=True).exists()
)
except UserConfig.DoesNotExist:
can_manage = False
can_cancel = False
can_manage_purchase = False
can_manage_stock = False
other_items_with_info = []
for oi in preparation_order.other_items.all():
other_items_with_info.append({
'item': oi,
'production_order': oi.production_orders.first(),
})
exit_batches = preparation_order.stock_batches.filter(movement_type='out').order_by('date')
return_batches = preparation_order.stock_batches.filter(movement_type='in').order_by('date')
# Discussion liée à ce bon de préparation (mobile)
discussion_messages_mobile = []
can_post_discussion_message_mobile = False
try:
from notifications.models import Discussion as _Discussion
from notifications.services import mark_discussion_seen as _mark_seen, can_post_preparation_order_message as _can_post
from django.contrib.contenttypes.models import ContentType as _CT
_ct = _CT.objects.get_for_model(preparation_order)
_discussion = _Discussion.objects.filter(content_type=_ct, object_id=preparation_order.pk).first()
if _discussion:
discussion_messages_mobile = list(
_discussion.messages.filter(is_deleted=False)
.select_related('author')
.prefetch_related('attachments')
.order_by('created_at')
)
_mark_seen(_discussion, user=request.user)
can_post_discussion_message_mobile = _can_post(request.user, preparation_order)
except Exception:
pass
return render(request, "stock/mobile/stock_preparation_order_detail_mobile.html", {
'preparation_order': preparation_order,
'items_with_stock': items_with_stock,
'other_items': other_items_with_info,
'can_manage': can_manage,
'can_cancel': can_cancel,
'can_manage_purchase': can_manage_purchase,
'can_manage_stock': can_manage_stock,
'exit_batches': exit_batches,
'return_batches': return_batches,
'has_executed_outputs': has_executed_outputs,
'is_reopened': is_reopened,
'deleted_items_to_return': deleted_items_to_return,
'discussion': _discussion,
'discussion_messages': discussion_messages_mobile,
'can_post_discussion_message': can_post_discussion_message_mobile,
})
@check_thematic_access('nature')
def green_surfaces_edit_standalone(request):
"""Mobile interface for editing green surfaces with CustomDrawToolbar in standalone mode (test page).
This page demonstrates the CustomDrawToolbar without callbacks.
The toolbar manages all drawing operations internally.
"""
context = {}
return render(request, "mobile/green_surfaces_edit_standalone.html", context)
@login_required
def mobile_assets_index(request):
"""
Dedicated mobile assets index page with MapLibre interactive map,
thematic and status filters, and switchable map/list view.
"""
import json
from common.models import Thematic, UserThematics, UserConfig
from assets.views.core import build_thematic_config, ASSET_STATUS_CHOICES
lang = get_language() or 'fr'
user_config = getattr(request.user, 'config', None)
if not user_config:
try:
user_config = UserConfig.objects.get(user=request.user)
except UserConfig.DoesNotExist:
user_config = None
if getattr(request.user, 'is_superuser', False):
accessible_thematics = list(Thematic.objects.order_by(f'{"name_nl" if lang == "nl" else "name_fr"}'))
elif user_config:
thematic_qs = (
UserThematics.objects
.filter(user_config=user_config, can_view_assets=True)
.select_related('thematic')
.order_by(f'thematic__{"name_nl" if lang == "nl" else "name_fr"}')
)
accessible_thematics = [tc.thematic for tc in thematic_qs]
else:
accessible_thematics = []
if not accessible_thematics and not getattr(request.user, 'is_superuser', False):
return render(request, "403.html", status=403)
thematic_config = build_thematic_config(accessible_thematics)
selected_thematic = request.GET.get('thematic', '')
if not selected_thematic and user_config and user_config.default_thematic and user_config.default_thematic in accessible_thematics:
selected_thematic = user_config.default_thematic.code
context = {
'accessible_thematics': accessible_thematics,
'thematic_config_json': json.dumps(thematic_config),
'asset_status_choices': list(ASSET_STATUS_CHOICES.items()),
'selected_thematic': selected_thematic,
'initial_query': request.GET.get('q', '').strip(),
'initial_view': request.GET.get('view', 'map'),
}
return render(request, "mobile/assets/assets_index_mobile.html", context)
@login_required
def mobile_assets_search(request):
"""
Search and filter endpoint for mobile assets.
Returns JSON results for list view and GeoJSON FeatureCollection for MapLibre map.
Supports proximity sorting if lat/lon parameters are provided.
"""
import math
import json
from urllib.parse import quote
from django.urls import reverse
from django.db.models import Q
from common.models import Thematic, UserThematics, UserConfig
from assets.models import (
NatureTree, NatureGreenSurface, NatureShrub, NatureBench, NatureRWIAsset,
SignPanel, SignPole,
TrafficLightPole, TrafficLightLantern, TrafficLightElectricalCabinet,
TrafficLightRadar, TrafficLightController,
PublicLightingPole, PublicLightingLightFixture, PublicLightingElectricalCabinet,
Roadway, RoadSidewalk, RoadFurniture, RoadDrain, RoadConcreteBlockSet,
CleanLitterBin, CleanGlassContainer,
ArtworkMonument, ArtworkFountain,
StructureGeoAsset, ControlCenterGeoAsset, ITSGeoAsset,
OutdoorAsset, ParkingSpot,
)
from assets.views.core import get_element_type_label
lang = get_language() or 'fr'
user_config = getattr(request.user, 'config', None)
if not user_config:
try:
user_config = UserConfig.objects.get(user=request.user)
except UserConfig.DoesNotExist:
user_config = None
if getattr(request.user, 'is_superuser', False):
accessible_thematics = list(Thematic.objects.all())
elif user_config:
thematic_qs = (
UserThematics.objects
.filter(user_config=user_config, can_view_assets=True)
.select_related('thematic')
)
accessible_thematics = [tc.thematic for tc in thematic_qs]
else:
accessible_thematics = []
accessible_codes = {t.code for t in accessible_thematics}
thematic_map = {t.code: t for t in accessible_thematics}
q = (request.GET.get('q') or '').strip()
thematic_filter = (request.GET.get('thematic') or '').strip()
status_filter = (request.GET.get('status') or '').strip()
MAX_MOBILE_ASSETS = 500
try:
limit = min(int(request.GET.get('limit', MAX_MOBILE_ASSETS)), MAX_MOBILE_ASSETS)
except (TypeError, ValueError):
limit = MAX_MOBILE_ASSETS
# L'utilisateur doit choisir au minimum une thématique pour voir quelque chose
if not thematic_filter or thematic_filter not in accessible_codes:
return JsonResponse({
'count': 0,
'total_count': 0,
'limit': limit,
'limit_reached': False,
'thematic': '',
'results': [],
'geojson': {
'type': 'FeatureCollection',
'features': [],
}
})
try:
user_lat = float(request.GET.get('lat'))
user_lon = float(request.GET.get('lon'))
except (TypeError, ValueError):
user_lat = None
user_lon = None
# Filtrage spatial par bounding box (bbox) sur la carte MapLibre
bbox_str = (request.GET.get('bbox') or '').strip()
bbox_poly_3812 = None
min_lon, min_lat, max_lon, max_lat = None, None, None, None
if bbox_str:
try:
parts = [float(x.strip()) for x in bbox_str.split(',')]
if len(parts) == 4:
min_lon, min_lat, max_lon, max_lat = parts
from django.contrib.gis.geos import Polygon
bbox_poly_4326 = Polygon.from_bbox((min_lon, min_lat, max_lon, max_lat))
bbox_poly_4326.srid = 4326
bbox_poly_3812 = bbox_poly_4326.clone()
bbox_poly_3812.transform(3812)
except Exception:
bbox_poly_3812 = None
min_lon, min_lat, max_lon, max_lat = None, None, None, None
thematic_models_registry = {
'nature': [
('naturetree', NatureTree),
('naturegreensurface', NatureGreenSurface),
('natureshrub', NatureShrub),
('naturebench', NatureBench),
('naturerwiasset', NatureRWIAsset),
],
'sign': [
('signpanel', SignPanel),
('signpole', SignPole),
],
'trafficlights': [
('trafficlightpole', TrafficLightPole),
('trafficlightradar', TrafficLightRadar),
('trafficlightcontroller', TrafficLightController),
('trafficlightlantern', TrafficLightLantern),
('trafficlightelectricalcabinet', TrafficLightElectricalCabinet),
],
'publiclighting': [
('publiclightingpole', PublicLightingPole),
('publiclightinglightfixture', PublicLightingLightFixture),
('publiclightingelectricalcabinet', PublicLightingElectricalCabinet),
],
'roads': [
('roadway', Roadway),
('roadsidewalk', RoadSidewalk),
('roadfurniture', RoadFurniture),
('roaddrain', RoadDrain),
('roadconcreteblockset', RoadConcreteBlockSet),
],
'clean': [
('cleanlitterbin', CleanLitterBin),
('cleanglasscontainer', CleanGlassContainer),
],
'artworks': [
('artworkmonument', ArtworkMonument),
('artworkfountain', ArtworkFountain),
],
'structures': [
('structuregeoasset', StructureGeoAsset),
],
'controlcenters': [
('controlcentergeoasset', ControlCenterGeoAsset),
],
'its': [
('itsgeoasset', ITSGeoAsset),
],
'outdoorfacilities': [
('outdoorasset', OutdoorAsset),
],
'outdoor': [
('outdoorasset', OutdoorAsset),
],
'parking': [
('parkingspot', ParkingSpot),
],
}
def _haversine(lat1, lon1, lat2, lon2):
R = 6371000
phi1 = math.radians(lat1)
phi2 = math.radians(lat2)
delta_phi = math.radians(lat2 - lat1)
delta_lambda = math.radians(lon2 - lon1)
a = math.sin(delta_phi / 2.0) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2.0) ** 2
c = 2.0 * math.atan2(math.sqrt(a), math.sqrt(max(0.0, 1.0 - a)))
return R * c
model_querysets = []
total_count = 0
t_code = thematic_filter
thematic_obj = thematic_map.get(t_code)
thematic_name = (getattr(thematic_obj, f'name_{lang}', None) or getattr(thematic_obj, 'name_fr', None) or t_code) if thematic_obj else t_code
thematic_icon = getattr(thematic_obj, 'icon', '') if thematic_obj else ''
models_list = thematic_models_registry.get(t_code, [])
for model_key, model_class in models_list:
qs = model_class.objects.all()
# Filtrage par boîte englobante (bbox / zoom de la carte)
if bbox_poly_3812:
if model_key == 'signpanel':
panel_geo_q = Q(geom__bboverlaps=bbox_poly_3812)
panel_geo_q |= Q(geom__isnull=True, pole__geom__bboverlaps=bbox_poly_3812)
panel_geo_q |= Q(geom__isnull=True, trafficlight_pole__geom__bboverlaps=bbox_poly_3812)
panel_geo_q |= Q(geom__isnull=True, publiclighting_pole__geom__bboverlaps=bbox_poly_3812)
if min_lon is not None:
panel_geo_q |= Q(lon__gte=min_lon, lon__lte=max_lon, lat__gte=min_lat, lat__lte=max_lat)
panel_geo_q |= Q(lon__isnull=True, pole__lon__gte=min_lon, pole__lon__lte=max_lon, pole__lat__gte=min_lat, pole__lat__lte=max_lat)
qs = qs.filter(panel_geo_q)
elif model_key == 'trafficlightlantern':
tl_geo_q = Q(geom__bboverlaps=bbox_poly_3812)
tl_geo_q |= Q(geom__isnull=True, pole__geom__bboverlaps=bbox_poly_3812)
if min_lon is not None:
tl_geo_q |= Q(lon__gte=min_lon, lon__lte=max_lon, lat__gte=min_lat, lat__lte=max_lat)
tl_geo_q |= Q(lon__isnull=True, pole__lon__gte=min_lon, pole__lon__lte=max_lon, pole__lat__gte=min_lat, pole__lat__lte=max_lat)
qs = qs.filter(tl_geo_q)
elif hasattr(model_class, 'geom'):
qs = qs.filter(geom__bboverlaps=bbox_poly_3812)
elif min_lon is not None and hasattr(model_class, 'lon') and hasattr(model_class, 'lat'):
qs = qs.filter(lon__gte=min_lon, lon__lte=max_lon, lat__gte=min_lat, lat__lte=max_lat)
if status_filter:
if hasattr(model_class, 'status'):
qs = qs.filter(status=status_filter)
else:
if hasattr(model_class, 'status'):
qs = qs.exclude(status='removed')
if q:
search_q = Q(code__icontains=q)
if hasattr(model_class, 'name_fr'):
search_q |= Q(name_fr__icontains=q)
if hasattr(model_class, 'name_nl'):
search_q |= Q(name_nl__icontains=q)
if model_key == 'signpanel' and hasattr(model_class, 'signpanel_type'):
search_q |= Q(signpanel_type__code__icontains=q)
search_q |= Q(signpanel_type__name_fr__icontains=q)
search_q |= Q(signpanel_type__name_nl__icontains=q)
qs = qs.filter(search_q)
# Nombre total d'assets correspondants (pour savoir si limite atteinte)
count_for_model = qs.count()
total_count += count_for_model
model_querysets.append((model_key, model_class, qs))
limit_reached = total_count > limit
# Si la limite est atteinte, ne renvoyer aucun asset ni GeoJSON pour ne pas induire en erreur et alléger la requête
if limit_reached:
return JsonResponse({
'count': 0,
'total_count': total_count,
'limit': limit,
'limit_reached': True,
'thematic': t_code,
'results': [],
'geojson': {
'type': 'FeatureCollection',
'features': [],
}
})
results = []
features = []
for model_key, model_class, qs in model_querysets:
if model_key == 'signpanel' and hasattr(model_class, 'signpanel_type'):
qs = qs.select_related('signpanel_type', 'model', 'pole', 'trafficlight_pole', 'publiclighting_pole')
elif model_key == 'trafficlightlantern' and hasattr(model_class, 'model'):
qs = qs.select_related('model', 'pole')
items = list(qs)
for item in items:
lat = None
lon = None
geometry = None
geom_wgs84 = None
# 1. Vérifier si un GeoJSON complet existe déjà sur l'asset (via GeoSyncMixin)
raw_geojson = getattr(item, 'geojson', None)
if raw_geojson:
if isinstance(raw_geojson, dict):
geometry = raw_geojson
elif isinstance(raw_geojson, str):
try:
geometry = json.loads(raw_geojson)
except Exception:
geometry = None
# 2. Vérifier si l'asset a un champ geom pour obtenir la géométrie 4326 et son centroïde
if hasattr(item, 'geom') and item.geom:
try:
geom_wgs84 = item.geom.clone()
if geom_wgs84.srid != 4326:
geom_wgs84.transform(4326)
if not geometry:
geometry = json.loads(geom_wgs84.geojson)
except Exception:
geom_wgs84 = None
# 3. Récupérer les coordonnées ponctuelles lon/lat (pour tri de distance et centroïde)
if hasattr(item, 'lat') and item.lat is not None and hasattr(item, 'lon') and item.lon is not None:
try:
lat = float(str(item.lat).replace(',', '.'))
lon = float(str(item.lon).replace(',', '.'))
except (ValueError, TypeError):
pass
if (lon is None or lat is None) and geom_wgs84:
try:
c = geom_wgs84.centroid
lon = float(c.x)
lat = float(c.y)
except Exception:
pass
if (lon is None or lat is None) and geometry and geometry.get('type') == 'Point' and geometry.get('coordinates'):
try:
lon = float(geometry['coordinates'][0])
lat = float(geometry['coordinates'][1])
except Exception:
pass
if (lon is None or lat is None) and hasattr(item, 'location') and item.location:
loc = item.location
if hasattr(loc, 'lat') and loc.lat and hasattr(loc, 'lon') and loc.lon:
try:
lat = float(str(loc.lat).replace(',', '.'))
lon = float(str(loc.lon).replace(',', '.'))
except (ValueError, TypeError):
pass
if (lon is None or lat is None):
for support_attr in ('pole', 'trafficlight_pole', 'publiclighting_pole'):
if hasattr(item, support_attr):
support = getattr(item, support_attr, None)
if support:
if hasattr(support, 'lat') and support.lat is not None and hasattr(support, 'lon') and support.lon is not None:
try:
lat = float(str(support.lat).replace(',', '.'))
lon = float(str(support.lon).replace(',', '.'))
break
except (ValueError, TypeError):
pass
if hasattr(support, 'geom') and support.geom:
try:
s_geom = support.geom.clone()
if s_geom.srid != 4326:
s_geom.transform(4326)
c = s_geom.centroid
lon = float(c.x)
lat = float(c.y)
break
except Exception:
pass
# 4. Fallback si geometry est absent mais qu'on a un point lon/lat
if not geometry and lon is not None and lat is not None:
geometry = {
'type': 'Point',
'coordinates': [lon, lat],
}
geom_type = geometry.get('type', 'Point') if geometry else 'Point'
# 5. Calcul de l'emprise bbox [min_lon, min_lat, max_lon, max_lat] pour zoom/cadrage
feature_bbox = None
if geom_wgs84 and geom_type != 'Point':
try:
ext = geom_wgs84.extent # (xmin, ymin, xmax, ymax)
feature_bbox = [float(ext[0]), float(ext[1]), float(ext[2]), float(ext[3])]
except Exception:
feature_bbox = None
distance_m = None
if user_lat is not None and user_lon is not None and lat is not None and lon is not None:
try:
distance_m = round(_haversine(user_lat, user_lon, lat, lon))
except Exception:
distance_m = None
item_name = (
getattr(item, f'name_{lang}', None) or
getattr(item, 'name_fr', None) or
(getattr(item.model, f'name_{lang}', None) or getattr(item.model, 'name_fr', None) if getattr(item, 'model', None) else '') or
''
)
signpanel_type = getattr(item, 'signpanel_type', None) if model_key == 'signpanel' else None
panel_type_code = getattr(signpanel_type, 'code', None) if signpanel_type else None
if panel_type_code:
panel_type_code = str(panel_type_code).strip()
elif model_key == 'signpanel':
raw_code = getattr(item, 'code', '') or ''
if re.match(r'^[A-Za-z]{1,3}\d{1,3}[A-Za-z]?$', raw_code.strip()):
panel_type_code = raw_code.strip()
svg_url = f"/static/assets/road_signs/2025/{panel_type_code.upper()}.svg" if panel_type_code else None
if model_key == 'signpanel':
if signpanel_type:
sign_name = getattr(signpanel_type, f'name_{lang}', None) or getattr(signpanel_type, 'name_fr', None) or ''
if not item_name:
item_name = f"{panel_type_code} - {sign_name}" if panel_type_code and sign_name else (sign_name or panel_type_code or '')
elif not item_name and panel_type_code:
item_name = panel_type_code
status_val = getattr(item, 'status', 'active')
status_display = item.get_status_display() if hasattr(item, 'get_status_display') else status_val
element_type = get_element_type_label(model_key)
if model_key == 'parkingspot' and hasattr(item, 'get_spot_type_display') and item.spot_type:
element_type = str(item.get_spot_type_display())
detail_url = reverse('mobile:mobile_asset_detail_direct', kwargs={'asset_model': model_key, 'asset_id': item.id})
spot_type = getattr(item, 'spot_type', None)
color_val = getattr(getattr(item, 'model', None), 'color', None) or getattr(item, 'color', None)
structure_type_val = getattr(item, 'structure_type', None)
feature_id = f"{model_key}_{item.id}"
pole_id = None
vertical_order = None
panel_width = None
panel_height = None
if model_key == 'signpanel':
pole_id = getattr(item, 'pole_id', None) or getattr(item, 'trafficlight_pole_id', None) or getattr(item, 'publiclighting_pole_id', None)
vertical_order = getattr(item, 'vertical_order', None)
try:
panel_width = item.get_width()
panel_height = item.get_height()
except Exception:
pass
elif hasattr(item, 'pole_id'):
pole_id = getattr(item, 'pole_id', None)
res_item = {
'id': item.id,
'feature_id': feature_id,
'code': getattr(item, 'code', f'#{item.id}'),
'name': item_name,
'thematic': t_code,
'thematic_name': thematic_name,
'thematic_icon': thematic_icon,
'model_key': model_key,
'asset_type': model_key,
'panel_type': panel_type_code,
'svg_url': svg_url,
'icon_url': svg_url,
'spot_type': spot_type,
'element_type': element_type,
'geom_type': geom_type,
'color': color_val,
'structure_type': structure_type_val,
'bbox': feature_bbox,
'status': status_val,
'status_display': status_display,
'lat': lat,
'lon': lon,
'distance_m': distance_m,
'detail_url': detail_url,
'pole_id': pole_id,
'vertical_order': vertical_order,
'panel_vertical_order': vertical_order,
'width': panel_width,
'height': panel_height,
}
results.append(res_item)
if geometry:
features.append({
'type': 'Feature',
'id': feature_id,
'geometry': geometry,
'bbox': feature_bbox,
'properties': {
'id': item.id,
'feature_id': feature_id,
'code': getattr(item, 'code', f'#{item.id}'),
'name': item_name,
'thematic': t_code,
'thematic_name': thematic_name,
'thematic_icon': thematic_icon,
'model_key': model_key,
'asset_type': model_key,
'panel_type': panel_type_code,
'svg_url': svg_url,
'icon_url': svg_url,
'spot_type': spot_type,
'element_type': element_type,
'geom_type': geom_type,
'color': color_val,
'structure_type': structure_type_val,
'bbox': feature_bbox,
'status': status_val,
'status_display': status_display,
'distance_m': distance_m,
'detail_url': detail_url,
'lon': lon,
'lat': lat,
'pole_id': pole_id,
'vertical_order': vertical_order,
'panel_vertical_order': vertical_order,
'width': panel_width,
'height': panel_height,
}
})
if user_lat is not None and user_lon is not None:
results.sort(key=lambda x: (x['distance_m'] if x['distance_m'] is not None else 999999999, x['code']))
else:
results.sort(key=lambda x: x['code'])
return JsonResponse({
'count': len(results),
'total_count': total_count,
'limit': limit,
'limit_reached': False,
'thematic': t_code,
'results': results,
'geojson': {
'type': 'FeatureCollection',
'features': features,
}
})