Valogeo/loko
Valogeo/loko
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add external_reference field to assets with permission support, improve intervention contract assignment logic, and sanitize template strings.

Browse source
This commit is contained in:
kdeterme 2026-08-26 16:52:09 +02:00
parent 4135c6ef47
commit 0b190526f8
9 changed files with 351 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
loko/assets/management/commands/sync_all_geojson.py Normal file
View file

@ -0,0 +1,85 @@
from django.core.management.base import BaseCommand
from django.apps import apps
from django.db import connection, transaction
class Command(BaseCommand):
help = "Synchronizes the geojson field (and lon/lat if missing) from geom for all assets/models using PostGIS SQL."
def add_arguments(self, parser):
parser.add_argument(
'--dry-run',
action='store_true',
help='Perform a dry run without modifying the database.',
)
parser.add_argument(
'--only-missing',
action='store_true',
help='Only update records where geojson is null or empty.',
)
def handle(self, *args, **options):
dry_run = options['dry_run']
only_missing = options['only_missing']
total_scanned = 0
total_updated = 0
with connection.cursor() as cursor:
for model in apps.get_models():
fields = [f.name for f in model._meta.get_fields()]
if 'geom' not in fields or 'geojson' not in fields:
continue
table = model._meta.db_table
is_json_field = model._meta.get_field('geojson').get_internal_type() == 'JSONField'
has_lon = 'lon' in fields
has_lat = 'lat' in fields
# Count total candidates with non-null geom
cursor.execute(f'SELECT COUNT(*) FROM "{table}" WHERE geom IS NOT NULL;')
count = cursor.fetchone()[0]
if count == 0:
continue
self.stdout.write(f"Processing {model._meta.app_label}.{model.__name__} ({table}, {count} candidate records)...")
if only_missing:
where_clause = "geom IS NOT NULL AND (geojson IS NULL OR geojson::text = '' OR geojson::text = 'null')"
else:
where_clause = "geom IS NOT NULL"
cursor.execute(f'SELECT COUNT(*) FROM "{table}" WHERE {where_clause};')
records_to_update = cursor.fetchone()[0]
if records_to_update == 0:
self.stdout.write(self.style.SUCCESS(f" {model.__name__}: 0 records need update."))
continue
if dry_run:
self.stdout.write(self.style.SUCCESS(f" [DRY-RUN] {model.__name__}: Would update {records_to_update}/{count} records."))
total_scanned += count
total_updated += records_to_update
continue
cast_geojson = "ST_AsGeoJSON(ST_Transform(geom, 4326))::jsonb" if is_json_field else "ST_AsGeoJSON(ST_Transform(geom, 4326))"
set_clauses = [f"geojson = {cast_geojson}"]
if has_lon:
set_clauses.append("lon = COALESCE(lon, ST_X(ST_PointOnSurface(ST_Transform(geom, 4326))))")
if has_lat:
set_clauses.append("lat = COALESCE(lat, ST_Y(ST_PointOnSurface(ST_Transform(geom, 4326))))")
set_str = ", ".join(set_clauses)
update_sql = f'UPDATE "{table}" SET {set_str} WHERE {where_clause};'
with transaction.atomic():
cursor.execute(update_sql)
updated_count = cursor.rowcount
self.stdout.write(self.style.SUCCESS(f" {model.__name__}: {updated_count}/{count} records updated in DB."))
total_scanned += count
total_updated += updated_count
prefix = "[DRY-RUN] Would update" if dry_run else "Successfully updated"
self.stdout.write(self.style.SUCCESS(f"\n{prefix} {total_updated} / {total_scanned} records across all models."))

1
loko/assets/migrations/0021_sync_asset_parent_data.py
View file

@ -10,6 +10,7 @@ def sync_parent_data(apps, schema_editor):
class Migration(migrations.Migration):
atomic = False
dependencies = [
('assets', '0020_trafficlightlantern_geojson_trafficlightlantern_geom_and_more'),

213
loko/assets/migrations/0025_artworkfountain_external_reference_and_more.py Normal file
View file

@ -0,0 +1,213 @@
# Generated by Django 6.0.7 on 2026-08-26 13:10
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('assets', '0024_structureexceptionaltransport_and_more'),
]
operations = [
migrations.AddField(
model_name='artworkfountain',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='artworkmonument',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='buildingasset',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='cleanglasscontainer',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='cleanlitterbin',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='controlcentergeoasset',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='itsgeoasset',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='naturebench',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='naturegreensurface',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='naturerwiasset',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='natureshrub',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='naturetree',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='outdoorasset',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='parkingspot',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='publiclightingelectricalcabinet',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='publiclightinglightfixture',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='publiclightingpole',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='roadconcreteblockset',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='roaddrain',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='roadfurniture',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='roadsidewalk',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='roadway',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='signpanel',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='signparkingspot',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='signpole',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='structuregeoasset',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightaccessory',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightcable',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightcontroller',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightcontrollerhardware',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightdetector',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightelectricalcabinet',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightelectricalcabinetcontent',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightlantern',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightnetworkhardware',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightpole',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightprogramming',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightradar',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightradarpole',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
migrations.AddField(
model_name='trafficlightvoltagehardware',
name='external_reference',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
),
]

1
loko/assets/models/core.py
View file

@ -268,6 +268,7 @@ class AbstractAsset(models.Model):
code = models.CharField(max_length=50)
name_fr = models.CharField(max_length=200,null=True,blank=True)
name_nl = models.CharField(max_length=200,null=True,blank=True)
external_reference = models.CharField(max_length=100, null=True, blank=True, verbose_name=_("Référence externe"))
status = models.CharField(choices=ASSET_STATUS_CHOICES, default='active', max_length=50)
obsolescence = models.CharField(choices=ASSET_OBSOLESCENCE_CHOICES, default='unknown', null=True, blank=True)
installation_date = models.DateTimeField(null=True,blank=True)

12
loko/assets/permissions.py
View file

@ -14,32 +14,32 @@ from common.models import UserConfig, UserThematics, UserAssetAccess, UserAssetT
# Fields editable by role for assets
ASSET_EDITABLE_FIELDS_BY_ROLE = {
'admin': [
'code', 'name_fr', 'name_nl', 'status', 'model', 'category',
'code', 'external_reference', 'name_fr', 'name_nl', 'status', 'model', 'category',
'serial_number', 'installation_date', 'brand', 'warranty_duration',
'last_inspection_date', 'funding_program', 'geom', 'lon', 'lat',
'location_id', 'intersection_id',
],
'manager': [
'code', 'name_fr', 'name_nl', 'status', 'model', 'category',
'code', 'external_reference', 'name_fr', 'name_nl', 'status', 'model', 'category',
'serial_number', 'installation_date', 'brand', 'warranty_duration',
'last_inspection_date', 'funding_program', 'geom', 'lon', 'lat',
'location_id', 'intersection_id',
],
'controller': [
'code', 'name_fr', 'name_nl', 'status', 'model', 'category',
'code', 'external_reference', 'name_fr', 'name_nl', 'status', 'model', 'category',
'serial_number', 'installation_date', 'brand', 'warranty_duration',
'last_inspection_date', 'funding_program', 'geom', 'lon', 'lat',
'location_id', 'intersection_id',
],
'external_manager': [
'name_fr', 'name_nl', 'category', 'brand', 'serial_number', 'funding_program'
'name_fr', 'name_nl', 'external_reference', 'category', 'brand', 'serial_number', 'funding_program'
],
'technician': ['name_fr', 'name_nl', 'category', 'brand', 'serial_number', 'funding_program'],
'technician': ['name_fr', 'name_nl', 'external_reference', 'category', 'brand', 'serial_number', 'funding_program'],
'operator': [],
'observer': [],
'viewer': [],
'editor': [
'code', 'name_fr', 'name_nl', 'status', 'model', 'category',
'code', 'external_reference', 'name_fr', 'name_nl', 'status', 'model', 'category',
'serial_number', 'installation_date', 'brand', 'warranty_duration',
'last_inspection_date', 'funding_program', 'geom', 'lon', 'lat',
'location_id', 'intersection_id',

15
loko/assets/templates/assets/detail_generic.html
View file

@ -223,6 +223,21 @@
</td>
</tr>
<!-- Référence externe -->
{% if asset.external_reference or 'external_reference' in allowed_update_fields %}
<tr class="asset-editable-field" data-field="external_reference" data-can-edit="{% if 'external_reference' in allowed_update_fields %}true{% else %}false{% endif %}">
<th>{% translate "Référence externe" %}</th>
<td>
<div class="field-display">
<span class="field-value">{{ asset.external_reference|default:"-" }}</span>
</div>
<div class="field-edit d-none">
<input type="text" class="form-control form-control-sm" value="{{ asset.external_reference|default:'' }}">
</div>
</td>
</tr>
{% endif %}
<!-- Nom FR -->
<tr class="asset-editable-field" data-field="name_fr" data-can-edit="{% if 'name_fr' in allowed_update_fields %}true{% else %}false{% endif %}">
<th>{% translate "Nom" %} (FR)</th>

2
loko/assets/templates/assets/structures_location_detail.html
View file

@ -908,7 +908,7 @@
}
})
.catch(err => {
alert('{% translate "Une erreur est survenue lors de l\'enregistrement." %}');
alert("{% filter escapejs %}{% translate "Une erreur est survenue lors de l'enregistrement." %}{% endfilter %}");
if (submitBtn) submitBtn.disabled = false;
if (spinner) spinner.classList.add('d-none');
});

6
loko/interventions/views/detail.py
View file

@ -1738,12 +1738,6 @@ def interventions_add(request, thematic_code):
intervention.save()
# Si aucun contrat encore défini via la commande, on utilise celui du symptôme
if not intervention.contract and intervention.symptom:
intervention.contract = intervention.symptom.contract
if intervention.symptom and not intervention.assigned_provider:
intervention.assigned_provider = intervention.symptom.provider
# Si le statut est "assigned" (pris en charge) et aucun membre n'est assigné, assigner l'utilisateur créateur et sa première équipe
if intervention.status == 'assigned' and not getattr(intervention, 'assigned_member_id', None):
from contracts.models import CompanyMember

49
loko/interventions/views/helpers.py
View file

@ -3269,15 +3269,7 @@ def resolve_intervention_contract_and_provider(thematic, symptom, asset=None, lo
thematic_code = thematic.code if hasattr(thematic, 'code') else (str(thematic) if thematic else None)
# 1. Contrat / prestataire direct du symptôme
if 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
# 2. Routage direct (thématique 'structures')
# 1. Routage direct (thématique 'structures') - prioritaire sur les contrats par défaut
if not contract and thematic_code == 'structures':
structure = None
if asset:
@ -3310,10 +3302,10 @@ def resolve_intervention_contract_and_provider(thematic, symptom, asset=None, lo
if not assigned_provider:
assigned_provider = contract.company
# 3. Contrat actif lié au maintenance_type du symptôme
if not contract and symptom and symptom.maintenance_type:
mt = symptom.maintenance_type
if thematic_code == 'trafficlights':
# 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):
@ -3351,17 +3343,26 @@ def resolve_intervention_contract_and_provider(thematic, symptom, asset=None, lo
structure_id = asset.structure_id
if structure_id:
struct_contract = StructureContract.objects.filter(
structure_id=structure_id,
maintenance_type=mt,
status='active'
).select_related('contract__company').first()
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':
elif thematic_code == 'clean' and mt:
clean_loc_id = None
if location:
if isinstance(location, CleanLocation):
@ -3385,7 +3386,7 @@ def resolve_intervention_contract_and_provider(thematic, symptom, asset=None, lo
if not assigned_provider:
assigned_provider = contract.company
elif thematic_code == 'its':
elif thematic_code == 'its' and mt:
its_contract = None
if asset and isinstance(asset, ITSGeoAsset):
its_contract = ITSGeoAssetContract.objects.filter(
@ -3405,6 +3406,14 @@ def resolve_intervention_contract_and_provider(thematic, symptom, asset=None, lo
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: