788 lines
30 KiB
Python
788 lines
30 KiB
Python
import json
|
||
import uuid
|
||
|
||
from django.contrib.auth.decorators import login_required
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from django.core.exceptions import PermissionDenied
|
||
from django.http import JsonResponse, Http404
|
||
from django.shortcuts import render, get_object_or_404
|
||
from django.utils.translation import gettext_lazy as _
|
||
from django.utils.translation import gettext
|
||
from django.views.decorators.http import require_GET, require_POST
|
||
|
||
from assets.models.roads import RoadConcreteBlockSet, CONCRETE_BLOCK_FUNCTION_CHOICES, CONCRETE_BLOCK_ACTION_CHOICES
|
||
from assets.models.core import ASSET_STATUS_CHOICES
|
||
from common.models import UserConfig
|
||
from common.private_files.helpers import build_private_url
|
||
from documents.models import DocumentAttachment
|
||
from projects.models import Project, STATUS_CHOICES as PROJECT_STATUS_CHOICES
|
||
from maps.permissions import map_access_required, check_map_access
|
||
from maps.models import MapConfig
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Maps – Index (landing page)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Maps accessible via the index page: code → URL name
|
||
MAP_URL_NAMES = {
|
||
'concrete_blocks': 'maps:concrete_blocks',
|
||
}
|
||
|
||
# Bootstrap icons per map code
|
||
MAP_ICONS = {
|
||
'concrete_blocks': 'bi-hexagon',
|
||
}
|
||
|
||
|
||
@login_required
|
||
def index(request):
|
||
"""Maps landing page — lists accessible maps as cards."""
|
||
user_config = get_object_or_404(UserConfig, user=request.user)
|
||
|
||
if not user_config.can_access_view('maps'):
|
||
raise PermissionDenied(gettext("Vous n'avez pas accès à cette vue."))
|
||
|
||
all_configs = MapConfig.objects.all()
|
||
accessible_maps = []
|
||
|
||
for config in all_configs:
|
||
result = check_map_access(request, config.code)
|
||
if result is True:
|
||
accessible_maps.append({
|
||
'config': config,
|
||
'url_name': MAP_URL_NAMES.get(config.code),
|
||
'icon': MAP_ICONS.get(config.code, 'bi-map'),
|
||
})
|
||
|
||
return render(request, 'maps/index.html', {
|
||
'accessible_maps': accessible_maps,
|
||
})
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Concrete Blocks – configuration
|
||
# ---------------------------------------------------------------------------
|
||
|
||
FUNCTION_LABELS = {
|
||
'protection': _('Protection'),
|
||
'antiterrorist': _('Antiterroriste'),
|
||
'parking_neutralization': _('Neutralisation stationnement'),
|
||
'road_safety': _('Sécurité routière'),
|
||
'public_space': _('Espace public'),
|
||
'construction': _('Chantier'),
|
||
'unidentified': _('Non identifié'),
|
||
'autre': _('Autre / Non renseigné'),
|
||
}
|
||
|
||
FUNCTION_COLORS = {
|
||
'protection': '#3498db',
|
||
'antiterrorist': '#e74c3c',
|
||
'parking_neutralization': "#f1509b",
|
||
'road_safety': '#f39c12',
|
||
'public_space': '#2ecc71',
|
||
'construction': '#9b59b6',
|
||
'unidentified': '#95a5a6',
|
||
'autre': '#7f8c8d',
|
||
}
|
||
|
||
RECOMMENDED_ACTION_LABELS = {
|
||
'keep': _('Maintenir'),
|
||
'remove': _('Enlever'),
|
||
'replace': _('Remplacer'),
|
||
}
|
||
|
||
STATUS_LABELS = {k: str(v) for k, v in ASSET_STATUS_CHOICES.items()}
|
||
|
||
|
||
|
||
|
||
def _get_cb_thematic_permissions(user):
|
||
"""Return (can_edit, can_create, can_delete) for the concrete-block asset type.
|
||
|
||
Priority order (mirrors _check_instance_permission):
|
||
1. Admin role → unconditional bypass.
|
||
2. UserAssetTypeAccess for RoadConcreteBlockSet → use its values directly.
|
||
(can_create is derived from can_edit since the model has no dedicated field.)
|
||
3. AssetTypePermissionConfig.requires_explicit_permissions=True → False.
|
||
4. Thematic 'roads' fallback (UserThematics + role-based actions).
|
||
"""
|
||
if not user or not user.is_authenticated:
|
||
return False, False, False
|
||
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from common.models import UserConfig, UserAssetTypeAccess, AssetTypePermissionConfig
|
||
|
||
try:
|
||
user_config = UserConfig.objects.get(user=user)
|
||
except UserConfig.DoesNotExist:
|
||
return False, False, False
|
||
|
||
# Step 1: admin bypass
|
||
if user_config.roles.filter(name='admin').exists():
|
||
return True, True, True
|
||
|
||
# Step 2: type-level explicit permissions for RoadConcreteBlockSet
|
||
ct = ContentType.objects.get_for_model(RoadConcreteBlockSet)
|
||
try:
|
||
type_access = UserAssetTypeAccess.objects.get(user_config=user_config, content_type=ct)
|
||
return type_access.can_edit, type_access.can_edit, type_access.can_delete
|
||
except UserAssetTypeAccess.DoesNotExist:
|
||
pass
|
||
|
||
# Step 3: type requires explicit permissions → deny all
|
||
try:
|
||
type_config = AssetTypePermissionConfig.objects.get(content_type=ct)
|
||
if type_config.requires_explicit_permissions:
|
||
return False, False, False
|
||
except AssetTypePermissionConfig.DoesNotExist:
|
||
pass
|
||
|
||
# Step 4: thematic fallback
|
||
from assets.permissions import get_allowed_actions_for_thematic
|
||
from common.models import Thematic
|
||
try:
|
||
thematic = Thematic.objects.get(code='roads')
|
||
actions = get_allowed_actions_for_thematic(user, thematic)
|
||
except Exception:
|
||
actions = []
|
||
return 'edit' in actions, 'create' in actions, 'archive' in actions
|
||
|
||
|
||
@map_access_required('concrete_blocks')
|
||
def concrete_blocks(request):
|
||
"""Map view for the concrete-block asset inventory."""
|
||
|
||
function_entries = [(k, str(FUNCTION_LABELS[k]), FUNCTION_COLORS[k]) for k in FUNCTION_LABELS]
|
||
can_edit, can_create, can_delete = _get_cb_thematic_permissions(request.user)
|
||
|
||
context = {
|
||
'function_entries': function_entries,
|
||
'function_colors_json': json.dumps(FUNCTION_COLORS),
|
||
'function_labels_json': json.dumps({k: str(v) for k, v in FUNCTION_LABELS.items()}),
|
||
'status_choices': [(k, str(v)) for k, v in ASSET_STATUS_CHOICES.items()],
|
||
'action_choices': [(k, str(v)) for k, v in RECOMMENDED_ACTION_LABELS.items()],
|
||
'function_choices': CONCRETE_BLOCK_FUNCTION_CHOICES,
|
||
'can_edit': can_edit,
|
||
'can_create': can_create,
|
||
'can_delete': can_delete,
|
||
}
|
||
return render(request, 'maps/concrete_blocks/index.html', context)
|
||
|
||
|
||
|
||
@map_access_required('concrete_blocks')
|
||
def concrete_blocks_geojson(request):
|
||
"""GeoJSON endpoint returning RoadConcreteBlockSet assets."""
|
||
|
||
filter_functions = request.GET.getlist('function')
|
||
filter_statuses = request.GET.getlist('status')
|
||
filter_actions = request.GET.getlist('action')
|
||
filter_min_blocks = request.GET.get('min_blocks', '')
|
||
filter_max_blocks = request.GET.get('max_blocks', '')
|
||
filter_min_priority = request.GET.get('min_priority', '')
|
||
filter_max_priority = request.GET.get('max_priority', '')
|
||
|
||
qs = (
|
||
RoadConcreteBlockSet.objects
|
||
.exclude(lon=None)
|
||
.exclude(lat=None)
|
||
.values(
|
||
'id', 'address', 'lon', 'lat',
|
||
'main_function', 'secondary_functions',
|
||
'number_of_blocks', 'targeted_action_date', 'status',
|
||
'priority_level', 'recommended_action', 'uninstallation_date',
|
||
'updated_at', 'observation_source_id',
|
||
)
|
||
)
|
||
|
||
features = []
|
||
for asset in qs:
|
||
main_fn = asset['main_function'] or 'autre'
|
||
secondary = asset['secondary_functions'] or []
|
||
|
||
# Build the full function list: main first, then secondary (deduplicated)
|
||
all_functions = [main_fn]
|
||
for fn in secondary:
|
||
if fn and fn != main_fn:
|
||
all_functions.append(fn)
|
||
|
||
primary_fn = all_functions[0]
|
||
|
||
if filter_functions:
|
||
if not any(f in filter_functions for f in all_functions):
|
||
continue
|
||
|
||
if filter_statuses and asset['status'] not in filter_statuses:
|
||
continue
|
||
|
||
if filter_actions and asset['recommended_action'] not in filter_actions:
|
||
continue
|
||
|
||
bc = asset['number_of_blocks']
|
||
if filter_min_blocks:
|
||
try:
|
||
if bc is None or bc < int(filter_min_blocks): continue
|
||
except ValueError:
|
||
pass
|
||
if filter_max_blocks:
|
||
try:
|
||
if bc is None or bc > int(filter_max_blocks): continue
|
||
except ValueError:
|
||
pass
|
||
|
||
pl = asset['priority_level']
|
||
if filter_min_priority:
|
||
try:
|
||
if pl is None or pl < int(filter_min_priority): continue
|
||
except ValueError:
|
||
pass
|
||
if filter_max_priority:
|
||
try:
|
||
if pl is None or pl > int(filter_max_priority): continue
|
||
except ValueError:
|
||
pass
|
||
|
||
functions_label = ', '.join(
|
||
str(FUNCTION_LABELS.get(f, f)) for f in all_functions
|
||
)
|
||
|
||
targeted_date = asset['targeted_action_date']
|
||
uninstall_date = asset['uninstallation_date']
|
||
features.append({
|
||
'type': 'Feature',
|
||
'geometry': {
|
||
'type': 'Point',
|
||
'coordinates': [asset['lon'], asset['lat']],
|
||
},
|
||
'properties': {
|
||
'asset_id': asset['id'],
|
||
'block_count': asset['number_of_blocks'],
|
||
'function': primary_fn,
|
||
'functions': all_functions,
|
||
'functions_key': 'cb:' + '|'.join(all_functions),
|
||
'functions_label': functions_label,
|
||
'address': asset['address'] or '',
|
||
'status': asset['status'] or '',
|
||
'targeted_action_date': targeted_date.strftime('%d/%m/%Y') if targeted_date else '',
|
||
'uninstallation_date': uninstall_date.strftime('%d/%m/%Y') if uninstall_date else '',
|
||
'priority_level': asset['priority_level'],
|
||
'recommended_action': str(RECOMMENDED_ACTION_LABELS.get(asset['recommended_action'], asset['recommended_action'] or '')),
|
||
'updated_at': asset['updated_at'].isoformat() if asset['updated_at'] else '',
|
||
'observation_source_id': asset['observation_source_id'],
|
||
},
|
||
})
|
||
|
||
total_blocks = int(sum(
|
||
f['properties']['block_count'] or 0
|
||
for f in features
|
||
if f['properties']['block_count'] is not None
|
||
))
|
||
return JsonResponse({'type': 'FeatureCollection', 'features': features, 'total_blocks': total_blocks})
|
||
|
||
|
||
@map_access_required('concrete_blocks')
|
||
def concrete_blocks_asset_photos(request, asset_id):
|
||
"""Return signed photo URLs for ManagedDocuments attached to a RoadConcreteBlockSet."""
|
||
try:
|
||
RoadConcreteBlockSet.objects.get(pk=asset_id)
|
||
except RoadConcreteBlockSet.DoesNotExist:
|
||
raise Http404
|
||
|
||
cbs_ct = ContentType.objects.get_for_model(RoadConcreteBlockSet)
|
||
attachments = (
|
||
DocumentAttachment.objects
|
||
.filter(content_type=cbs_ct, object_id=asset_id)
|
||
.select_related('document', 'document__current_version')
|
||
)
|
||
|
||
photos = []
|
||
for att in attachments:
|
||
if not att.document:
|
||
continue
|
||
version = att.document.current_version
|
||
if version is None or not version.file or not version.file.name:
|
||
continue
|
||
has_thumb = bool(version.thumbnail and version.thumbnail.name)
|
||
photos.append({
|
||
'thumb_url': (
|
||
build_private_url('documents', 'DocumentVersion', version.pk, 'thumbnail')
|
||
if has_thumb
|
||
else build_private_url('documents', 'DocumentVersion', version.pk, 'file')
|
||
),
|
||
'full_url': build_private_url('documents', 'DocumentVersion', version.pk, 'file'),
|
||
'title': att.document.title or '',
|
||
})
|
||
|
||
return JsonResponse({'photos': photos})
|
||
|
||
|
||
@map_access_required('concrete_blocks')
|
||
def projects_new_jersey_geojson(request):
|
||
"""GeoJSON endpoint returning Project polygons tagged 'new-jersey'."""
|
||
qs = Project.objects.filter(tags__slug='new-jersey').exclude(geom=None)
|
||
|
||
features = []
|
||
for project in qs:
|
||
geom_4326 = project.geom.transform(4326, clone=True)
|
||
features.append({
|
||
'type': 'Feature',
|
||
'geometry': json.loads(geom_4326.geojson),
|
||
'properties': {
|
||
'project_id': project.id,
|
||
'code': project.code,
|
||
'name': project.name,
|
||
'status_key': project.status or '',
|
||
'status': str(PROJECT_STATUS_CHOICES.get(project.status, project.status or '')),
|
||
},
|
||
})
|
||
|
||
return JsonResponse({'type': 'FeatureCollection', 'features': features})
|
||
|
||
|
||
@map_access_required('concrete_blocks')
|
||
def projects_new_jersey_list(request):
|
||
"""JSON list of projects tagged 'new-jersey' for the card panel."""
|
||
qs = Project.objects.filter(tags__slug='new-jersey').order_by('-code')
|
||
|
||
projects = []
|
||
for project in qs:
|
||
bbox = None
|
||
if project.geom:
|
||
geom_4326 = project.geom.transform(4326, clone=True)
|
||
bbox = list(geom_4326.extent) # [xmin, ymin, xmax, ymax]
|
||
|
||
projects.append({
|
||
'id': project.id,
|
||
'code': project.code,
|
||
'name': project.name,
|
||
'description': project.description or '',
|
||
'pmo_number': project.pmo_number,
|
||
'osiris_number': project.osiris_number,
|
||
'status_key': project.status or '',
|
||
'status': str(PROJECT_STATUS_CHOICES.get(project.status, project.status or '')),
|
||
'start_date': project.start_date.strftime('%d/%m/%Y') if project.start_date else '',
|
||
'start_date_iso': project.start_date.isoformat() if project.start_date else '',
|
||
'end_date': project.end_date.strftime('%d/%m/%Y') if project.end_date else '',
|
||
'end_date_iso': project.end_date.isoformat() if project.end_date else '',
|
||
'has_geom': project.geom is not None,
|
||
'bbox': bbox,
|
||
'url': project.get_absolute_url(),
|
||
})
|
||
|
||
return JsonResponse({'projects': projects})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Concrete Blocks – CRUD API
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _apply_concrete_block_data(asset, data, user):
|
||
"""
|
||
Apply field updates to a RoadConcreteBlockSet instance and save it.
|
||
Returns (success: bool, error_message: str | None).
|
||
"""
|
||
from datetime import date
|
||
from django.contrib.gis.geos import Point
|
||
|
||
try:
|
||
if 'address' in data:
|
||
asset.address = data['address'] or None
|
||
if 'status' in data and data['status'] in ASSET_STATUS_CHOICES:
|
||
asset.status = data['status']
|
||
if 'main_function' in data:
|
||
asset.main_function = data['main_function'] or None
|
||
if 'secondary_functions' in data:
|
||
v = data['secondary_functions']
|
||
asset.secondary_functions = list(v) if v else None
|
||
if 'number_of_blocks' in data:
|
||
v = data['number_of_blocks']
|
||
asset.number_of_blocks = int(v) if v not in (None, '') else None
|
||
if 'length' in data:
|
||
v = data['length']
|
||
asset.length = float(str(v).replace(',', '.')) if v not in (None, '') else None
|
||
if 'recommended_action' in data:
|
||
asset.recommended_action = data['recommended_action'] or None
|
||
if 'priority_level' in data:
|
||
v = data['priority_level']
|
||
asset.priority_level = int(v) if v not in (None, '') else None
|
||
if 'targeted_action_date' in data:
|
||
v = data['targeted_action_date']
|
||
asset.targeted_action_date = date.fromisoformat(str(v)) if v else None
|
||
if 'uninstallation_date' in data:
|
||
v = data['uninstallation_date']
|
||
asset.uninstallation_date = date.fromisoformat(str(v)) if v else None
|
||
|
||
# Position: update both lon/lat and 3812 geometry together
|
||
if 'lon' in data and 'lat' in data:
|
||
lon_v, lat_v = data['lon'], data['lat']
|
||
if lon_v not in (None, '') and lat_v not in (None, ''):
|
||
asset.lon = float(lon_v)
|
||
asset.lat = float(lat_v)
|
||
pt = Point(asset.lon, asset.lat, srid=4326)
|
||
pt.transform(3812)
|
||
asset.geom = pt
|
||
else:
|
||
asset.lon = None
|
||
asset.lat = None
|
||
asset.geom = None
|
||
|
||
if hasattr(asset, '_history_user'):
|
||
asset._history_user = user
|
||
asset.save()
|
||
return True, None
|
||
except Exception as exc:
|
||
return False, str(exc)
|
||
|
||
|
||
@require_GET
|
||
@map_access_required('concrete_blocks')
|
||
def concrete_blocks_asset_detail_api(request, asset_id):
|
||
"""Return full asset data for the edit form."""
|
||
try:
|
||
asset = RoadConcreteBlockSet.objects.get(pk=asset_id)
|
||
except RoadConcreteBlockSet.DoesNotExist:
|
||
raise Http404
|
||
|
||
from assets.permissions import can_edit_asset, can_delete_asset
|
||
|
||
return JsonResponse({
|
||
'id': asset.id,
|
||
'code': asset.code or '',
|
||
'address': asset.address or '',
|
||
'status': asset.status or 'active',
|
||
'main_function': asset.main_function or '',
|
||
'secondary_functions': asset.secondary_functions or [],
|
||
'number_of_blocks': asset.number_of_blocks,
|
||
'length': asset.length,
|
||
'recommended_action': asset.recommended_action or '',
|
||
'priority_level': asset.priority_level,
|
||
'targeted_action_date': asset.targeted_action_date.strftime('%Y-%m-%d') if asset.targeted_action_date else '',
|
||
'uninstallation_date': asset.uninstallation_date.strftime('%Y-%m-%d') if asset.uninstallation_date else '',
|
||
'lon': asset.lon,
|
||
'lat': asset.lat,
|
||
'observation_source_id': asset.observation_source_id,
|
||
'can_edit': can_edit_asset(request.user, asset),
|
||
'can_delete': can_delete_asset(request.user, asset),
|
||
})
|
||
|
||
|
||
@require_POST
|
||
@map_access_required('concrete_blocks')
|
||
def concrete_blocks_asset_update_api(request, asset_id):
|
||
"""Update a single concrete block asset."""
|
||
try:
|
||
asset = RoadConcreteBlockSet.objects.get(pk=asset_id)
|
||
except RoadConcreteBlockSet.DoesNotExist:
|
||
raise Http404
|
||
|
||
from assets.permissions import can_edit_asset
|
||
if not can_edit_asset(request.user, asset):
|
||
return JsonResponse({'success': False, 'error': gettext('Permission refusée')}, status=403)
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
except json.JSONDecodeError:
|
||
return JsonResponse({'success': False, 'error': gettext('JSON invalide')}, status=400)
|
||
|
||
success, error = _apply_concrete_block_data(asset, data, request.user)
|
||
if not success:
|
||
return JsonResponse({'success': False, 'error': error}, status=400)
|
||
return JsonResponse({'success': True, 'id': asset.id})
|
||
|
||
|
||
@require_POST
|
||
@map_access_required('concrete_blocks')
|
||
def concrete_blocks_asset_create_api(request):
|
||
"""Create a new concrete block asset."""
|
||
_, can_create, _ = _get_cb_thematic_permissions(request.user)
|
||
if not can_create:
|
||
return JsonResponse({'success': False, 'error': gettext('Permission refusée')}, status=403)
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
except json.JSONDecodeError:
|
||
return JsonResponse({'success': False, 'error': gettext('JSON invalide')}, status=400)
|
||
|
||
asset = RoadConcreteBlockSet()
|
||
code = (data.pop('code', None) or '').strip()
|
||
if not code:
|
||
# Auto-generate a unique code
|
||
prefix = 'NJB'
|
||
import datetime
|
||
suffix = datetime.date.today().strftime('%Y%m%d')
|
||
base = f'{prefix}-{suffix}'
|
||
n = RoadConcreteBlockSet.objects.filter(code__startswith=base).count() + 1
|
||
code = f'{base}-{n:03d}'
|
||
asset.code = code
|
||
asset.status = 'active'
|
||
|
||
success, error = _apply_concrete_block_data(asset, data, request.user)
|
||
if not success:
|
||
return JsonResponse({'success': False, 'error': error}, status=400)
|
||
return JsonResponse({'success': True, 'id': asset.id, 'code': asset.code})
|
||
|
||
|
||
@require_POST
|
||
@map_access_required('concrete_blocks')
|
||
def concrete_blocks_asset_delete_api(request, asset_id):
|
||
"""Archive (soft-delete) a concrete block asset."""
|
||
try:
|
||
asset = RoadConcreteBlockSet.objects.get(pk=asset_id)
|
||
except RoadConcreteBlockSet.DoesNotExist:
|
||
raise Http404
|
||
|
||
from assets.permissions import can_delete_asset
|
||
if not can_delete_asset(request.user, asset):
|
||
return JsonResponse({'success': False, 'error': gettext('Permission refusée')}, status=403)
|
||
|
||
from django.utils import timezone
|
||
asset.status = 'removed'
|
||
asset.uninstallation_date = timezone.now()
|
||
if hasattr(asset, '_history_user'):
|
||
asset._history_user = request.user
|
||
asset.save(update_fields=['status', 'uninstallation_date'])
|
||
return JsonResponse({'success': True})
|
||
|
||
|
||
@require_POST
|
||
@map_access_required('concrete_blocks')
|
||
def concrete_blocks_bulk_update_api(request):
|
||
"""Bulk-update multiple concrete block assets with the same field values."""
|
||
|
||
from assets.permissions import can_edit_asset
|
||
|
||
try:
|
||
data = json.loads(request.body)
|
||
except json.JSONDecodeError:
|
||
return JsonResponse({'success': False, 'error': gettext('JSON invalide')}, status=400)
|
||
|
||
asset_ids = data.get('asset_ids', [])
|
||
updates = data.get('updates', {})
|
||
|
||
if not asset_ids:
|
||
return JsonResponse({'success': False, 'error': gettext('Aucun asset sélectionné')}, status=400)
|
||
if not updates:
|
||
return JsonResponse({'success': False, 'error': gettext('Aucune modification spécifiée')}, status=400)
|
||
|
||
assets = list(RoadConcreteBlockSet.objects.filter(pk__in=asset_ids))
|
||
updated, errors = 0, []
|
||
|
||
for asset in assets:
|
||
if not can_edit_asset(request.user, asset):
|
||
errors.append(asset.id)
|
||
continue
|
||
ok, _ = _apply_concrete_block_data(asset, updates, request.user)
|
||
if ok:
|
||
updated += 1
|
||
else:
|
||
errors.append(asset.id)
|
||
|
||
return JsonResponse({'success': True, 'updated': updated, 'errors': errors})
|
||
|
||
|
||
@require_GET
|
||
@login_required
|
||
@map_access_required('concrete_blocks')
|
||
def projects_new_jersey_export_excel(request):
|
||
"""Export filtered New-Jersey projects as an Excel file with their custom attributes."""
|
||
import io
|
||
from datetime import date
|
||
from django.db.models import Q
|
||
from django.http import HttpResponse
|
||
from django.utils.translation import gettext as _
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
|
||
from openpyxl.utils import get_column_letter
|
||
|
||
from projects.permissions import filter_viewable_projects_for_user
|
||
from projects.models import ProjectTemplate, ProjectTemplateAttribute, STATUS_CHOICES as PROJECT_STATUS_CHOICES
|
||
|
||
# 1. Fetch user-viewable projects and filter by tag 'new-jersey'
|
||
projects_qs = filter_viewable_projects_for_user(request.user).filter(tags__slug='new-jersey').order_by('-code')
|
||
|
||
# 2. Apply search filter
|
||
search_query = (request.GET.get('q') or '').strip()
|
||
if search_query:
|
||
q_filter = Q(code__icontains=search_query) | Q(name__icontains=search_query) | Q(description__icontains=search_query)
|
||
try:
|
||
val = int(search_query)
|
||
q_filter |= Q(pmo_number=val) | Q(osiris_number=val)
|
||
except ValueError:
|
||
pass
|
||
projects_qs = projects_qs.filter(q_filter)
|
||
|
||
# 3. Apply status filter
|
||
statuses = request.GET.getlist('status')
|
||
if statuses:
|
||
projects_qs = projects_qs.filter(status__in=statuses)
|
||
|
||
# 4. Apply date filters
|
||
date_from = (request.GET.get('date_from') or '').strip()
|
||
if date_from:
|
||
projects_qs = projects_qs.filter(Q(end_date__isnull=True) | Q(end_date__gte=date_from))
|
||
|
||
date_to = (request.GET.get('date_to') or '').strip()
|
||
if date_to:
|
||
projects_qs = projects_qs.filter(Q(start_date__isnull=True) | Q(start_date__lte=date_to))
|
||
|
||
# 5. Prefetch related fields to optimize DB queries
|
||
projects_list = list(projects_qs.prefetch_related(
|
||
'templates__attributes',
|
||
'attribute_values__attribute__choice_set__choices',
|
||
'thematics',
|
||
'tags'
|
||
).select_related('created_by'))
|
||
|
||
# 6. Gather all unique template attributes present in the exported projects
|
||
attributes = []
|
||
seen_attr_ids = set()
|
||
for project in projects_list:
|
||
for template in project.templates.all():
|
||
for attr in template.attributes.all():
|
||
if attr.id not in seen_attr_ids:
|
||
seen_attr_ids.add(attr.id)
|
||
attributes.append(attr)
|
||
|
||
# Sort attributes to make sure the columns have a stable order
|
||
attributes.sort(key=lambda a: (a.template.name_fr, a.order, a.name_fr))
|
||
|
||
# Map project values for easy lookup
|
||
values_map = {}
|
||
for project in projects_list:
|
||
project_values = {}
|
||
for val in project.attribute_values.all():
|
||
project_values[val.attribute_id] = val.display_value()
|
||
values_map[project.id] = project_values
|
||
|
||
# 7. Build Workbook
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = str(_("Projets New-Jersey"))
|
||
|
||
# Explicitly show grid lines
|
||
ws.views.sheetView[0].showGridLines = True
|
||
|
||
# Styling colors
|
||
primary_color = "1F4E79" # Deep navy blue
|
||
text_color = "FFFFFF"
|
||
zebra_color = "F9FAFB" # Off-white / light grey
|
||
border_color = "D9D9D9" # Light gray
|
||
|
||
# Styles definition
|
||
header_font = Font(name="Calibri", size=11, bold=True, color=text_color)
|
||
header_fill = PatternFill(start_color=primary_color, end_color=primary_color, fill_type="solid")
|
||
header_alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
|
||
|
||
data_font = Font(name="Calibri", size=10)
|
||
data_alignment_left = Alignment(horizontal="left", vertical="center")
|
||
data_alignment_center = Alignment(horizontal="center", vertical="center")
|
||
data_alignment_right = Alignment(horizontal="right", vertical="center")
|
||
|
||
thin_border = Border(
|
||
left=Side(style='thin', color=border_color),
|
||
right=Side(style='thin', color=border_color),
|
||
top=Side(style='thin', color=border_color),
|
||
bottom=Side(style='thin', color=border_color)
|
||
)
|
||
|
||
zebra_fill = PatternFill(start_color=zebra_color, end_color=zebra_color, fill_type="solid")
|
||
|
||
def get_alignment_for_attr(attr_type):
|
||
if attr_type == 'number':
|
||
return data_alignment_right
|
||
elif attr_type in ('date', 'boolean'):
|
||
return data_alignment_center
|
||
return data_alignment_left
|
||
|
||
# Headers definition
|
||
fixed_headers = [
|
||
str(_("Code")),
|
||
str(_("Nom")),
|
||
str(_("Description")),
|
||
str(_("Statut")),
|
||
str(_("Numéro PMO")),
|
||
str(_("Numéro OSIRIS")),
|
||
str(_("Date de début")),
|
||
str(_("Date de fin")),
|
||
str(_("Créé par")),
|
||
str(_("Thématiques")),
|
||
str(_("Tags")),
|
||
]
|
||
|
||
attribute_headers = [f"{attr.template.name} — {attr.name}" for attr in attributes]
|
||
all_headers = fixed_headers + attribute_headers
|
||
|
||
# Write header row
|
||
for col_idx, header in enumerate(all_headers, start=1):
|
||
cell = ws.cell(row=1, column=col_idx, value=header)
|
||
cell.font = header_font
|
||
cell.fill = header_fill
|
||
cell.alignment = header_alignment
|
||
cell.border = thin_border
|
||
|
||
ws.row_dimensions[1].height = 28
|
||
|
||
# Write data rows
|
||
for row_idx, project in enumerate(projects_list, start=2):
|
||
ws.row_dimensions[row_idx].height = 20
|
||
use_zebra = (row_idx % 2 == 1)
|
||
|
||
row_data = [
|
||
(project.code or "", data_alignment_center),
|
||
(project.name or "", data_alignment_left),
|
||
(project.description or "", data_alignment_left),
|
||
(str(PROJECT_STATUS_CHOICES.get(project.status, project.status or "")), data_alignment_center),
|
||
(project.pmo_number if project.pmo_number is not None else "", data_alignment_center),
|
||
(project.osiris_number if project.osiris_number is not None else "", data_alignment_center),
|
||
(project.start_date.strftime("%d/%m/%Y") if project.start_date else "", data_alignment_center),
|
||
(project.end_date.strftime("%d/%m/%Y") if project.end_date else "", data_alignment_center),
|
||
(project.created_by.get_full_name() or project.created_by.username if project.created_by else "", data_alignment_left),
|
||
(", ".join(t.name for t in project.thematics.all()), data_alignment_left),
|
||
(", ".join(t.name for t in project.tags.all()), data_alignment_left),
|
||
]
|
||
|
||
# Dynamic attributes
|
||
for attr in attributes:
|
||
val_str = values_map[project.id].get(attr.id, "")
|
||
align = get_alignment_for_attr(attr.attribute_type)
|
||
row_data.append((val_str, align))
|
||
|
||
for col_idx, (val, align) in enumerate(row_data, start=1):
|
||
cell = ws.cell(row=row_idx, column=col_idx, value=val)
|
||
cell.font = data_font
|
||
cell.alignment = align
|
||
cell.border = thin_border
|
||
if use_zebra:
|
||
cell.fill = zebra_fill
|
||
|
||
# Auto-filter on header
|
||
if all_headers:
|
||
ws.auto_filter.ref = f"A1:{get_column_letter(len(all_headers))}1"
|
||
|
||
# Freeze panes
|
||
ws.freeze_panes = "A2"
|
||
|
||
# Auto-adjust column widths
|
||
for col_idx, header in enumerate(all_headers, start=1):
|
||
col_letter = get_column_letter(col_idx)
|
||
max_length = len(header)
|
||
for row in range(2, len(projects_list) + 2):
|
||
cell_val = ws.cell(row=row, column=col_idx).value
|
||
if cell_val:
|
||
max_length = max(max_length, len(str(cell_val)))
|
||
ws.column_dimensions[col_letter].width = min(max(max_length + 3, 12), 40)
|
||
|
||
# Stream response
|
||
output = io.BytesIO()
|
||
wb.save(output)
|
||
output.seek(0)
|
||
|
||
filename = f"projets_new_jersey_{date.today().strftime('%Y%m%d')}.xlsx"
|
||
response = HttpResponse(
|
||
output.read(),
|
||
content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
)
|
||
response["Content-Disposition"] = f'attachment; filename="{filename}"'
|
||
return response
|