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

refactor: remove entire schools application module and clean up related configurations

Browse source
This commit is contained in:
kdeterme 2026-07-26 14:59:20 +02:00
parent 034cf1e918
commit f3207e946c
29 changed files with 3 additions and 2676 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
streetup/common/context_processors.py
View file

@ -53,18 +53,18 @@ def current_app_view(request):
Context processor that determines the current active AppView based on the URL namespace. Context processor that determines the current active AppView based on the URL namespace.
Makes 'current_view' available in all templates. Makes 'current_view' available in all templates.
Only 'sign', 'documents', and 'schools' have their own dedicated views. Only 'sign' and 'documents' have their own dedicated views.
All other apps (assets, interventions, controls, contracts, etc.) use 'asset_management' view. All other apps (assets, interventions, controls, contracts, etc.) use 'asset_management' view.
""" """
current_view = None current_view = None
# Apps that have their own dedicated view # Apps that have their own dedicated view
DEDICATED_VIEW_APPS = ['sign', 'documents', 'schools', 'reporting', 'maps', 'authorizations', 'panoramax'] DEDICATED_VIEW_APPS = ['sign', 'documents', 'reporting', 'maps', 'authorizations', 'panoramax']
try: try:
# Get the namespace from the current URL # Get the namespace from the current URL
resolved = resolve(request.path) resolved = resolve(request.path)
namespace = resolved.namespace # e.g., 'schools', 'sign', 'home', 'assets' namespace = resolved.namespace # e.g., 'sign', 'home', 'assets'
if namespace: if namespace:
if namespace in DEDICATED_VIEW_APPS: if namespace in DEDICATED_VIEW_APPS:

13
streetup/common/fixtures/reference_data.json
View file

@ -178,19 +178,6 @@
"name": "observer" "name": "observer"
} }
}, },
{
"model": "common.appview",
"pk": 1,
"fields": {
"code": "schools",
"name_fr": "Écoles",
"name_nl": "Scholen",
"icon": "bi-building",
"url_name": "schools:index",
"is_default": false,
"order": 30
}
},
{ {
"model": "common.appview", "model": "common.appview",
"pk": 2, "pk": 2,

0
streetup/schools/__init__.py
View file

10
streetup/schools/admin.py
View file

@ -1,10 +0,0 @@
from django.contrib import admin
from .models import GeocodingTask
@admin.register(GeocodingTask)
class GeocodingTaskAdmin(admin.ModelAdmin):
list_display = ['id', 'user', 'status', 'created_at', 'completed_at', 'total_rows', 'successful_geocodes']
list_filter = ['status', 'created_at']
search_fields = ['user__username', 'user__email']
readonly_fields = ['created_at', 'started_at', 'completed_at']

6
streetup/schools/apps.py
View file

@ -1,6 +0,0 @@
from django.apps import AppConfig
class SchoolsConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'schools'

35
streetup/schools/forms.py
View file

@ -1,35 +0,0 @@
from django import forms
from .models import GeocodingTask
class GeocodingUploadForm(forms.ModelForm):
"""Form for uploading Excel files for geocoding"""
class Meta:
model = GeocodingTask
fields = ['uploaded_file']
widgets = {
'uploaded_file': forms.FileInput(attrs={
'class': 'form-control',
'accept': '.xls,.xlsx',
})
}
labels = {
'uploaded_file': 'Fichier Excel'
}
help_texts = {
'uploaded_file': 'Sélectionnez un fichier Excel (.xls ou .xlsx) contenant les adresses à géolocaliser'
}
def clean_uploaded_file(self):
file = self.cleaned_data.get('uploaded_file')
if file:
# Check file extension
if not file.name.lower().endswith(('.xls', '.xlsx')):
raise forms.ValidationError('Le fichier doit être au format Excel (.xls ou .xlsx)')
# Check file size (max 10MB)
if file.size > 10 * 1024 * 1024:
raise forms.ValidationError('Le fichier ne doit pas dépasser 10 Mo')
return file

443
streetup/schools/geocoding.py
View file

@ -1,443 +0,0 @@
# geocode_photon.py
# ------------------------------------------------------------
# Geocode a list of addresses from an Excel file using Photon.
#
# Based on the original geocode_photon.py script provided.
# Adapted for use within Django application.
# Now with internal database geocoding using fuzzy matching as first attempt,
# with fallback to Photon API.
# ------------------------------------------------------------
import asyncio
import aiohttp
import math
import time
import logging
from typing import Any, Dict, Optional, Tuple, List, Union
import pandas as pd
from .internal_geocoding import geocode_internal, geocode_internal_batch, DEFAULT_SCORE_THRESHOLD
logger = logging.getLogger(__name__)
PHOTON_URL = "https://photon.komoot.io/api/"
def chunked(seq: List[Any], size: int):
for i in range(0, len(seq), size):
yield seq[i:i+size]
def _preview(q: str, n: int = 80) -> str:
q = q.replace("\n", " ").replace("\r", " ").strip()
return (q if len(q) <= n else q[: n - 1] + "…")
def _format_found_address(properties: Dict[str, Any]) -> str:
"""Format the address from Photon response properties into a readable string."""
parts = []
# House number and street
street = properties.get("street", "")
housenumber = properties.get("housenumber", "")
if street:
if housenumber:
parts.append(f"{street} {housenumber}")
else:
parts.append(street)
elif properties.get("name"):
parts.append(properties.get("name"))
# Postcode and city
postcode = properties.get("postcode", "")
city = properties.get("city", "") or properties.get("locality", "") or properties.get("district", "")
if postcode and city:
parts.append(f"{postcode} {city}")
elif city:
parts.append(city)
elif postcode:
parts.append(postcode)
# State (if different from city)
state = properties.get("state", "")
if state and state != city:
parts.append(state)
# Country
country = properties.get("country", "")
if country:
parts.append(country)
return ", ".join(parts) if parts else properties.get("name", "Unknown")
async def fetch_photon(
session: aiohttp.ClientSession,
query: str,
limit: int = 1,
timeout_s: int = 10,
max_retries: int = 4,
) -> Tuple[Optional[float], Optional[float], str]:
"""Call Photon for a single query and return (lon, lat, found_address_or_error).
Returns:
Tuple of (longitude, latitude, found_address_or_error)
- On success: (lon, lat, formatted_address)
- On not found: (None, None, "Not found")
- On error: (None, None, error_description)
"""
params = {"q": query, "limit": str(limit)}
attempts = 0
last_http: Optional[int] = None
last_error: Optional[str] = None
t0 = time.perf_counter()
backoff = 1.0
for attempt in range(max_retries):
attempts += 1
try:
async with session.get(PHOTON_URL, params=params, timeout=timeout_s) as resp:
last_http = resp.status
# Retry on rate limit or server errors
if resp.status in (429, 500, 502, 503, 504):
await resp.read() # drain
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 16)
continue
resp.raise_for_status()
data = await resp.json()
features = data.get("features") or []
if not features:
elapsed = time.perf_counter() - t0
logger.debug(f"[req] '{_preview(query)}' -> not_found (http={last_http}) "
f"in {elapsed:.2f}s, attempts={attempts}")
return (None, None, "Not found")
# Take the first feature
feature = features[0]
geom = feature.get("geometry") or {}
properties = feature.get("properties") or {}
coords = geom.get("coordinates") or []
if isinstance(coords, list) and len(coords) >= 2:
lon, lat = coords[0], coords[1]
if all(isinstance(v, (int, float)) and math.isfinite(v) for v in (lon, lat)):
found_address = _format_found_address(properties)
elapsed = time.perf_counter() - t0
logger.debug(f"[req] '{_preview(query)}' -> success lon={lon}, lat={lat} "
f"(http={last_http}) in {elapsed:.2f}s, attempts={attempts}")
return (float(lon), float(lat), found_address)
# If we reached here the response is 200 but unusable
elapsed = time.perf_counter() - t0
logger.warning(f"[req] '{_preview(query)}' -> invalid_geometry (http={last_http}) "
f"in {elapsed:.2f}s, attempts={attempts}")
return (None, None, "Invalid geometry in response")
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
last_error = e.__class__.__name__
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 16)
# loop to retry
# Exhausted retries
elapsed = time.perf_counter() - t0
if last_error:
logger.warning(f"[req] '{_preview(query)}' -> error {last_error} "
f"after {elapsed:.2f}s, attempts={attempts}")
return (None, None, f"Error: {last_error}")
else:
logger.warning(f"[req] '{_preview(query)}' -> failed (http={last_http}) "
f"after {elapsed:.2f}s, attempts={attempts}")
return (None, None, f"Failed (HTTP {last_http})")
async def fetch_with_internal_fallback(
session: aiohttp.ClientSession,
query: str,
limit: int = 1,
timeout_s: int = 10,
max_retries: int = 4,
internal_threshold: float = DEFAULT_SCORE_THRESHOLD,
use_internal: bool = True,
) -> Tuple[Optional[float], Optional[float], str]:
"""
Try to geocode using internal database first, then fall back to Photon API.
Args:
session: aiohttp session for Photon API calls
query: Address string to geocode
limit: Photon limit parameter
timeout_s: Timeout for Photon API calls
max_retries: Max retries for Photon API calls
internal_threshold: Score threshold for internal geocoding (0-100)
use_internal: Whether to try internal DB first (default: True)
Returns:
Tuple of (longitude, latitude, found_address_or_source)
The found_address includes source info: "[Internal] address" or "[Photon] address"
"""
if not query or not query.strip():
return (None, None, "Empty address")
# Try internal geocoding first if enabled
if use_internal:
try:
# Run synchronous database query in a thread pool to avoid blocking the event loop
loop = asyncio.get_event_loop()
lon, lat, found_addr, score = await loop.run_in_executor(
None, geocode_internal, query, internal_threshold
)
if lon is not None and lat is not None:
# Success with internal DB
logger.debug(f"Internal DB success for '{_preview(query)}' (score: {score:.1f})")
return (lon, lat, f"[URBIS] {found_addr}")
else:
logger.debug(f"Internal DB no match for '{_preview(query)}', falling back to Photon")
except Exception as e:
logger.warning(f"Internal geocoding error for '{_preview(query)}': {e}, falling back to Photon")
# Fall back to Photon API
lon, lat, found_addr = await fetch_photon(session, query, limit, timeout_s, max_retries)
if lon is not None and lat is not None:
return (lon, lat, f"[Photon API] {found_addr}")
return (lon, lat, found_addr)
async def geocode_block(
session: aiohttp.ClientSession,
rows: List[Tuple[int, str]],
limit: int = 1,
internal_threshold: float = DEFAULT_SCORE_THRESHOLD,
use_internal: bool = True,
internal_concurrency: int = 5,
) -> Dict[int, Tuple[Optional[float], Optional[float], str]]:
"""Geocode a block (list) of (row_index, address) concurrently.
Uses a two-phase approach:
1. First, try internal DB geocoding in parallel for all addresses
2. Then, fallback to Photon API only for addresses that failed internal geocoding
Args:
session: aiohttp session for API calls
rows: List of (row_index, address) tuples
limit: Photon API limit parameter
internal_threshold: Score threshold for internal geocoding
use_internal: Whether to try internal DB first
internal_concurrency: Max concurrent internal DB queries (default 5)
Returns:
Dict mapping row_index to (lon, lat, found_address_or_error)
"""
out: Dict[int, Tuple[Optional[float], Optional[float], str]] = {}
if not rows:
return out
# Phase 1: Try internal geocoding for all addresses in parallel
need_photon: List[Tuple[int, str]] = []
if use_internal:
try:
internal_results = await geocode_internal_batch(
rows,
threshold=internal_threshold,
max_concurrent=internal_concurrency
)
for idx, addr in rows:
result = internal_results.get(idx)
if result:
lon, lat, found_addr, score = result
if lon is not None and lat is not None:
# Success with internal DB
out[idx] = (lon, lat, f"[URBIS] {found_addr}")
logger.debug(f"Internal DB success for idx={idx} (score: {score:.2f})")
else:
# Need to try Photon
need_photon.append((idx, addr))
else:
need_photon.append((idx, addr))
except Exception as e:
logger.warning(f"Internal batch geocoding error: {e}, falling back to Photon for all")
need_photon = list(rows)
else:
need_photon = list(rows)
# Phase 2: Fallback to Photon API for addresses that failed internal geocoding
if need_photon:
logger.debug(f"Falling back to Photon for {len(need_photon)} addresses")
photon_tasks = [
fetch_photon(session, addr, limit=limit)
for (idx, addr) in need_photon
]
photon_results = await asyncio.gather(*photon_tasks, return_exceptions=True)
for (idx, _), res in zip(need_photon, photon_results):
if isinstance(res, Exception):
out[idx] = (None, None, f"Error: {res.__class__.__name__}")
elif res is None:
out[idx] = (None, None, "Unknown error")
else:
lon, lat, found_addr = res
if lon is not None and lat is not None:
out[idx] = (lon, lat, f"[Photon API] {found_addr}")
else:
out[idx] = (lon, lat, found_addr)
return out
async def geocode_all(
addresses: List[str],
chunk_size: int = 5,
limit: int = 1,
internal_threshold: float = DEFAULT_SCORE_THRESHOLD,
use_internal: bool = True,
) -> List[Tuple[Optional[float], Optional[float], str]]:
"""Geocode all addresses in blocks of `chunk_size`.
First tries internal database with fuzzy matching, then falls back to Photon API.
Args:
addresses: List of address strings to geocode
chunk_size: Number of concurrent requests per block
limit: Photon API limit parameter
internal_threshold: Score threshold for internal geocoding (0-100)
use_internal: Whether to try internal DB first (default: True)
Returns:
List of (lon, lat, found_address_or_error) aligned with input order."""
connector = aiohttp.TCPConnector(limit=chunk_size) # matches concurrency per block
headers = {
"User-Agent": "streetup-geocoder/1.0 (+https://photon.komoot.io/)",
"Accept": "application/json",
}
total = len(addresses)
total_blocks = (total + max(1, chunk_size) - 1) // max(1, chunk_size)
async with aiohttp.ClientSession(connector=connector, headers=headers) as session:
results: Dict[int, Tuple[Optional[float], Optional[float], str]] = {}
# Prepare indexed rows
indexed = [(i, (a or "").strip()) for i, a in enumerate(addresses)]
for block_idx, block in enumerate(chunked(indexed, chunk_size), start=1):
t0 = time.perf_counter()
# Skip empty addresses quickly
non_empty = [(i, a) for i, a in block if a]
empty_count = len(block) - len(non_empty)
if not non_empty:
for i, _ in block:
results[i] = (None, None, "Empty address")
took = time.perf_counter() - t0
logger.info(f"[{block_idx}/{total_blocks}] Skipped empty block of {len(block)} rows (all empty). Took {took:.2f}s.")
continue
block_results = await geocode_block(
session, non_empty, limit=limit,
internal_threshold=internal_threshold,
use_internal=use_internal
)
for i, a in block:
if not a:
results[i] = (None, None, "Empty address")
else:
results[i] = block_results.get(i, (None, None, "Unknown error"))
successes = sum(1 for i, _ in non_empty if results.get(i, (None, None, ""))[0] is not None)
failures = len(non_empty) - successes
took = time.perf_counter() - t0
logger.info(f"[{block_idx}/{total_blocks}] Completed block of {len(block)} rows: {successes} success, {failures} not found/errors, {empty_count} empty. Took {took:.2f}s.")
return [results[i] for i in range(len(addresses))]
def resolve_address_series(df: pd.DataFrame, address_col: Union[int, str]) -> pd.Series:
if isinstance(address_col, int):
if address_col < 0 or address_col >= df.shape[1]:
raise IndexError(f"Address column index {address_col} out of range.")
return df.iloc[:, address_col]
else:
if address_col not in df.columns:
raise KeyError(f"Column '{address_col}' not found. Available: {list(df.columns)}")
return df[address_col]
def geocode_excel_file(
input_path: str,
output_path: str,
sheet_name: Optional[str] = None,
address_col: Union[int, str] = 0,
chunk_size: int = 3,
limit: int = 1,
output_sheet: str = "geocoded",
internal_threshold: float = DEFAULT_SCORE_THRESHOLD,
use_internal: bool = True,
) -> dict:
"""
Geocode addresses from an Excel file and write results to a new Excel file.
First tries internal database with fuzzy matching in both FR and NL,
then falls back to Photon API if score is below threshold.
Args:
input_path: Path to input Excel file
output_path: Path to output Excel file
sheet_name: Sheet name to read (None for first sheet)
address_col: Address column (index like 0 or column name)
chunk_size: Concurrent requests per block (default 3)
limit: Photon 'limit' parameter (default 1)
output_sheet: Name of the output sheet (default "geocoded")
internal_threshold: Score threshold for internal geocoding (0-100, default 80)
use_internal: Whether to try internal DB first (default True)
Returns:
dict: Statistics about the geocoding operation
"""
# Load Excel
if sheet_name:
df = pd.read_excel(input_path, sheet_name=sheet_name)
else:
df = pd.read_excel(input_path)
# Resolve address column
try:
# Try to parse address-col as int; if fails use as name
try:
addr_col: Union[int, str] = int(address_col)
except (ValueError, TypeError):
addr_col = str(address_col)
s = resolve_address_series(df, addr_col)
except (IndexError, KeyError) as e:
raise ValueError(f"Error: {e}")
addresses = s.astype(str).fillna("").tolist()
logger.info(f"Starting geocoding of {len(addresses)} rows in blocks of {chunk_size}...")
t0_all = time.perf_counter()
# Run async geocoding in blocks
results = asyncio.run(geocode_all(
addresses, chunk_size=chunk_size, limit=limit,
internal_threshold=internal_threshold,
use_internal=use_internal
))
lons = [r[0] for r in results]
lats = [r[1] for r in results]
found_addresses = [r[2] for r in results]
took_all = time.perf_counter() - t0_all
total = len(addresses)
total_ok = sum(1 for r in results if r[0] is not None and r[1] is not None)
total_empty = sum(1 for a in addresses if not a.strip())
total_fail = total - total_ok - total_empty
logger.info(f"Finished geocoding: {total_ok} success, {total_fail} not found/errors, {total_empty} empty. Took {took_all:.2f}s.")
# Attach results
df_out = df.copy()
df_out["longitude"] = lons
df_out["latitude"] = lats
df_out["found_address"] = found_addresses
# Write to Excel
with pd.ExcelWriter(output_path, engine="openpyxl") as writer:
df_out.to_excel(writer, index=False, sheet_name=output_sheet)
logger.info(f"Done. Wrote results to sheet '{output_sheet}' in {output_path}.")
return {
'total': total,
'success': total_ok,
'failed': total_fail,
'empty': total_empty,
'duration': took_all,
}

344
streetup/schools/internal_geocoding.py
View file

@ -1,344 +0,0 @@
# internal_geocoding.py
# ------------------------------------------------------------
# Internal geocoding using PostgreSQL pg_trgm fuzzy matching
# Searches AdrPoint and AdrStreet models using trigram similarity
# Requires: CREATE EXTENSION pg_trgm;
# Recommended: GiST indexes with gist_trgm_ops for optimal ORDER BY <-> performance
# ------------------------------------------------------------
import logging
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Optional, Tuple, List, Dict
from django.db import connection
from common.models import AdrPoint, AdrStreet
logger = logging.getLogger(__name__)
# Thread pool for parallel DB queries - limit to avoid connection pool exhaustion
_db_executor = ThreadPoolExecutor(max_workers=5)
# Default score threshold (0.0 to 100.0) - addresses with score below this will fallback to Photon
# pg_trgm similarity scores range from 0.0 to 1.0 (scaled to 0-100 in results)
DEFAULT_SCORE_THRESHOLD = 30.0
def normalize_address(address: str) -> str:
"""Normalize an address string for better matching"""
if not address:
return ""
# Convert to lowercase and strip whitespace
address = address.lower().strip()
# Remove extra spaces
address = " ".join(address.split())
return address
def search_address_points(query: str, limit: int = 5) -> List[Tuple[AdrPoint, float]]:
"""
Search for address points in the database using PostgreSQL trigram distance.
Uses the <-> operator which is optimized for ORDER BY LIMIT queries.
Runs two separate queries (FR and NL) and merges results for optimal index usage.
Args:
query: Address query string
limit: Maximum number of results to return
Returns:
List of tuples (AdrPoint, score) ordered by score descending
"""
normalized_query = normalize_address(query)
if not normalized_query:
return []
# Run two separate queries for FR and NL to properly use GiST indexes
# Using LEAST() on two columns prevents index usage
sql_fr = """
SELECT id, 1 - (full_address_fr <-> %s) as score
FROM common_adrpoint
WHERE lon IS NOT NULL AND lat IS NOT NULL AND full_address_fr IS NOT NULL
ORDER BY full_address_fr <-> %s
LIMIT %s;
"""
sql_nl = """
SELECT id, 1 - (full_address_nl <-> %s) as score
FROM common_adrpoint
WHERE lon IS NOT NULL AND lat IS NOT NULL AND full_address_nl IS NOT NULL
ORDER BY full_address_nl <-> %s
LIMIT %s;
"""
with connection.cursor() as cursor:
# Query FR
cursor.execute(sql_fr, [normalized_query, normalized_query, limit])
rows_fr = cursor.fetchall()
# Query NL
cursor.execute(sql_nl, [normalized_query, normalized_query, limit])
rows_nl = cursor.fetchall()
# Merge results, keeping best score per id (scaling pg_trgm scores to 0-100)
scores_by_id: Dict[int, float] = {}
for row in rows_fr:
scores_by_id[row[0]] = row[1] * 100.0
for row in rows_nl:
val = row[1] * 100.0
if row[0] in scores_by_id:
scores_by_id[row[0]] = max(scores_by_id[row[0]], val)
else:
scores_by_id[row[0]] = val
if not scores_by_id:
return []
# Get top N by score
top_ids = sorted(scores_by_id.keys(), key=lambda x: scores_by_id[x], reverse=True)[:limit]
# Fetch the actual AdrPoint objects
points = list(AdrPoint.objects.select_related(
'street_name', 'municipality'
).filter(id__in=top_ids))
# Build results maintaining the score order
results = [(point, scores_by_id[point.id]) for point in points]
results.sort(key=lambda x: x[1], reverse=True)
return results
def search_streets(query: str, limit: int = 5) -> List[Tuple[AdrStreet, float]]:
"""
Search for streets in the database using PostgreSQL trigram distance.
Uses the <-> operator which is optimized for ORDER BY LIMIT queries.
Runs two separate queries (FR and NL) and merges results for optimal index usage.
Args:
query: Street query string
limit: Maximum number of results to return
Returns:
List of tuples (AdrStreet, score) ordered by score descending
"""
normalized_query = normalize_address(query)
if not normalized_query:
return []
# Run two separate queries for FR and NL to properly use GiST indexes
sql_fr = """
SELECT id, 1 - (full_street_fr <-> %s) as score
FROM common_adrstreet
WHERE lon IS NOT NULL AND lat IS NOT NULL AND full_street_fr IS NOT NULL
ORDER BY full_street_fr <-> %s
LIMIT %s;
"""
sql_nl = """
SELECT id, 1 - (full_street_nl <-> %s) as score
FROM common_adrstreet
WHERE lon IS NOT NULL AND lat IS NOT NULL AND full_street_nl IS NOT NULL
ORDER BY full_street_nl <-> %s
LIMIT %s;
"""
with connection.cursor() as cursor:
# Query FR
cursor.execute(sql_fr, [normalized_query, normalized_query, limit])
rows_fr = cursor.fetchall()
# Query NL
cursor.execute(sql_nl, [normalized_query, normalized_query, limit])
rows_nl = cursor.fetchall()
# Merge results, keeping best score per id (scaling pg_trgm scores to 0-100)
scores_by_id: Dict[int, float] = {}
for row in rows_fr:
scores_by_id[row[0]] = row[1] * 100.0
for row in rows_nl:
val = row[1] * 100.0
if row[0] in scores_by_id:
scores_by_id[row[0]] = max(scores_by_id[row[0]], val)
else:
scores_by_id[row[0]] = val
if not scores_by_id:
return []
# Get top N by score
top_ids = sorted(scores_by_id.keys(), key=lambda x: scores_by_id[x], reverse=True)[:limit]
# Fetch the actual AdrStreet objects
streets = list(AdrStreet.objects.select_related('municipality').filter(id__in=top_ids))
# Build results maintaining the score order
results = [(street, scores_by_id[street.id]) for street in streets]
results.sort(key=lambda x: x[1], reverse=True)
return results
def search_streets_in_memory(query: str, limit: int = 5) -> List[Tuple[AdrStreet, float]]:
"""
Alternative: Search streets using in-memory fuzzy matching with rapidfuzz.
Useful when pg_trgm is not available or for small datasets (~3000 streets).
Args:
query: Street query string
limit: Maximum number of results to return
Returns:
List of tuples (AdrStreet, score) ordered by score descending
"""
try:
from rapidfuzz import fuzz
except ImportError:
logger.warning("rapidfuzz not installed, falling back to pg_trgm")
return search_streets(query, limit)
normalized_query = normalize_address(query)
if not normalized_query:
return []
# Load all streets in memory (only ~3000 records)
streets = AdrStreet.objects.select_related('municipality').filter(
lon__isnull=False,
lat__isnull=False,
full_street_fr__isnull=False
)
results = []
for street in streets:
# Calculate scores using rapidfuzz (already on 0-100 scale)
score_fr = fuzz.token_set_ratio(normalized_query, street.full_street_fr.lower() if street.full_street_fr else "")
score_nl = fuzz.token_set_ratio(normalized_query, street.full_street_nl.lower() if street.full_street_nl else "")
best_score = max(score_fr, score_nl)
if best_score > 10.0:
results.append((street, best_score))
# Sort by score descending and return top results
results.sort(key=lambda x: x[1], reverse=True)
return results[:limit]
def geocode_internal(
address: str,
threshold: float = DEFAULT_SCORE_THRESHOLD
) -> Tuple[Optional[float], Optional[float], str, float]:
"""
Try to geocode an address using the internal database with PostgreSQL trigram matching.
Args:
address: Address string to geocode
threshold: Minimum score threshold (0.0-1.0) to accept a match
Returns:
Tuple of (longitude, latitude, found_address, score)
- On success: (lon, lat, formatted_address, score)
- On no match: (None, None, "No match in internal DB", 0)
"""
if not address or not address.strip():
return (None, None, "Empty address", 0)
# First, try to find address points (more precise)
point_results = search_address_points(address, limit=1)
if point_results:
point, score = point_results[0]
if score >= threshold:
# Use the pre-computed full address or build it
found_address = point.full_address_fr or ""
if not found_address:
street_name = point.street_name.name_fr if point.street_name else ""
municipality_name = point.municipality.name if point.municipality else ""
found_address = f"{point.number} {street_name}"
if municipality_name:
found_address = f"{found_address}, {municipality_name}"
logger.debug(f"Found address point: '{address}' -> '{found_address}' (score: {score:.2f})")
return (point.lon, point.lat, found_address, score)
# If no address point found, try streets
street_results = search_streets(address, limit=1)
if street_results:
street, score = street_results[0]
if score >= threshold:
# Use the pre-computed full street name or build it
found_address = street.full_street_fr or ""
if not found_address:
street_name = street.name_fr
municipality_name = street.municipality.name if street.municipality else ""
found_address = street_name
if municipality_name:
found_address = f"{found_address}, {municipality_name}"
logger.debug(f"Found street: '{address}' -> '{found_address}' (score: {score:.2f})")
return (street.lon, street.lat, found_address, score)
# No match found above threshold
all_scores = [s for _, s in point_results + street_results]
best_score_found = max(all_scores, default=0)
logger.debug(f"No internal match for '{address}' (best score: {best_score_found:.2f})")
return (None, None, "No match in internal DB", 0)
async def geocode_internal_batch(
addresses: List[Tuple[int, str]],
threshold: float = DEFAULT_SCORE_THRESHOLD,
max_concurrent: int = 5,
) -> Dict[int, Tuple[Optional[float], Optional[float], str, float]]:
"""
Geocode multiple addresses in parallel using the internal database.
Args:
addresses: List of (index, address) tuples
threshold: Minimum score threshold (0.0-1.0) to accept a match
max_concurrent: Maximum concurrent DB queries (default 5)
Returns:
Dict mapping index to (longitude, latitude, found_address, score)
"""
if not addresses:
return {}
loop = asyncio.get_event_loop()
# Create tasks for parallel execution
async def geocode_one(idx: int, addr: str) -> Tuple[int, Tuple[Optional[float], Optional[float], str, float]]:
"""Geocode a single address and return (index, result)"""
try:
result = await loop.run_in_executor(
_db_executor,
geocode_internal,
addr,
threshold
)
return (idx, result)
except Exception as e:
logger.warning(f"Internal geocoding error for index {idx}: {e}")
return (idx, (None, None, f"Error: {e}", 0))
# Process in chunks to limit concurrent DB connections
results: Dict[int, Tuple[Optional[float], Optional[float], str, float]] = {}
for i in range(0, len(addresses), max_concurrent):
chunk = addresses[i:i + max_concurrent]
tasks = [geocode_one(idx, addr) for idx, addr in chunk]
chunk_results = await asyncio.gather(*tasks, return_exceptions=True)
for res in chunk_results:
if isinstance(res, Exception):
logger.warning(f"Unexpected error in batch geocoding: {res}")
else:
idx, geocode_result = res
results[idx] = geocode_result
return results

BIN
streetup/schools/locale/en/LC_MESSAGES/django.mo
View file

Binary file not shown.

197
streetup/schools/locale/en/LC_MESSAGES/django.po
View file

@ -1,197 +0,0 @@
# English translations for the Schools module.
# Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER
# This file is distributed under the same license as the PACKAGE package.
#
msgid ""
msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2026-05-23 14:00+0200\n"
"PO-Revision-Date: 2025-12-04 13:00+0100\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: English <LL@li.org>\n"
"Language: en\n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=UTF-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Plural-Forms: nplurals=2; plural=(n != 1);\n"
#: schools/templates/schools/base.html:7
msgid "Accueil"
msgstr "Home"
#: schools/templates/schools/base.html:11
#: schools/templates/schools/index.html:19
msgid "Géolocalisation"
msgstr "Geolocation"
#: schools/templates/schools/geolocation/index.html:11
msgid "Géolocalisation d'adresses"
msgstr "Address geolocation"
#: schools/templates/schools/geolocation/index.html:14
msgid ""
"Uploadez un fichier Excel contenant des adresses dans la première colonne "
"pour les géolocaliser automatiquement."
msgstr ""
"Upload an Excel file containing addresses in the first column to geolocate "
"them automatically."
#: schools/templates/schools/geolocation/index.html:36
msgid "Uploader un fichier"
msgstr "Upload a file"
#: schools/templates/schools/geolocation/index.html:60
msgid "Format du fichier :"
msgstr "File format:"
#: schools/templates/schools/geolocation/index.html:62
msgid "Le fichier doit être au format Excel (.xls ou .xlsx)"
msgstr "The file must be in Excel format (.xls or .xlsx)"
#: schools/templates/schools/geolocation/index.html:63
msgid "Les adresses doivent être dans la première colonne"
msgstr "Addresses must be in the first column"
#: schools/templates/schools/geolocation/index.html:64
msgid "Taille maximale : 10 Mo"
msgstr "Maximum size: 10 MB"
#: schools/templates/schools/geolocation/index.html:69
msgid "Lancer la géolocalisation"
msgstr "Start geolocation"
#: schools/templates/schools/geolocation/index.html:77
msgid "Comment ça fonctionne ?"
msgstr "How does it work?"
#: schools/templates/schools/geolocation/index.html:81
msgid "Préparez un fichier Excel avec les adresses dans la première colonne"
msgstr "Prepare an Excel file with addresses in the first column"
#: schools/templates/schools/geolocation/index.html:82
msgid "Uploadez le fichier en utilisant le formulaire ci-dessus"
msgstr "Upload the file using the form above"
#: schools/templates/schools/geolocation/index.html:83
msgid "Le système géolocalise automatiquement chaque adresse"
msgstr "The system automatically geolocates each address"
#: schools/templates/schools/geolocation/index.html:84
msgid ""
"Vous recevrez un email avec le fichier résultat contenant les coordonnées "
"(longitude et latitude)"
msgstr ""
"You will receive an email with the result file containing the coordinates "
"(longitude and latitude)"
#: schools/templates/schools/geolocation/index.html:88
msgid ""
"Le traitement peut prendre plusieurs minutes selon le nombre d'adresses."
msgstr ""
"Processing may take several minutes depending on the number of addresses."
#: schools/templates/schools/geolocation/index.html:97
msgid "Mes tâches de géolocalisation"
msgstr "My geolocation tasks"
#: schools/templates/schools/geolocation/index.html:105
msgid "Fichier"
msgstr "File"
#: schools/templates/schools/geolocation/index.html:106
msgid "Date"
msgstr "Date"
#: schools/templates/schools/geolocation/index.html:107
msgid "Statut"
msgstr "Status"
#: schools/templates/schools/geolocation/index.html:108
msgid "Adresses"
msgstr "Addresses"
#: schools/templates/schools/geolocation/index.html:109
msgid "Réussies"
msgstr "Successful"
#: schools/templates/schools/geolocation/index.html:110
msgid "Action"
msgstr "Action"
#: schools/templates/schools/geolocation/index.html:122
#: schools/templates/schools/geolocation/index.html:181
msgid "En attente"
msgstr "Pending"
#: schools/templates/schools/geolocation/index.html:124
#: schools/templates/schools/geolocation/index.html:182
msgid "En cours"
msgstr "In progress"
#: schools/templates/schools/geolocation/index.html:126
#: schools/templates/schools/geolocation/index.html:183
msgid "Terminé"
msgstr "Completed"
#: schools/templates/schools/geolocation/index.html:128
#: schools/templates/schools/geolocation/index.html:184
msgid "Échoué"
msgstr "Failed"
#: schools/templates/schools/geolocation/index.html:136
#: schools/templates/schools/geolocation/index.html:193
msgid "Télécharger"
msgstr "Download"
#: schools/templates/schools/geolocation/index.html:140
#: schools/templates/schools/geolocation/index.html:197
msgid "Chargement..."
msgstr "Loading..."
#: schools/templates/schools/geolocation/index.html:158
msgid "Aucune tâche de géolocalisation pour le moment."
msgstr "No geolocation task at the moment."
#: schools/templates/schools/index.html:9
msgid "Écoles"
msgstr "Schools"
#: schools/templates/schools/index.html:10
msgid "Bienvenue dans la gestion des écoles"
msgstr "Welcome to school management"
#: schools/templates/schools/index.html:22
msgid ""
"Uploadez un fichier Excel contenant des adresses pour les géolocaliser "
"automatiquement."
msgstr ""
"Upload an Excel file containing addresses to geolocate them automatically."
#: schools/templates/schools/index.html:25
msgid "Accéder"
msgstr "Access"
#: schools/views.py:33 schools/views.py:51
msgid "Vous n'avez pas accès à cette vue."
msgstr "You do not have access to this view."
#: schools/views.py:73
msgid ""
"Votre fichier a été uploadé avec succès. La géolocalisation est en cours. "
"Vous recevrez un email lorsque le traitement sera terminé."
msgstr ""
"Your file has been uploaded successfully. Geolocation is in progress. You "
"will receive an email when processing is complete."
#: schools/views.py:268
msgid "Vous n'avez pas accès à ce fichier."
msgstr "You do not have access to this file."
#: schools/views.py:272
msgid "Le fichier résultat n'est pas disponible."
msgstr "The result file is not available."
#: schools/views.py:276
msgid "Le fichier résultat n'a pas été trouvé."
msgstr "The result file was not found."

BIN
streetup/schools/locale/fr/LC_MESSAGES/django.mo
View file

Binary file not shown.

198
streetup/schools/locale/fr/LC_MESSAGES/django.po
View file

@ -1,198 +0,0 @@
# Traductions françaises pour le module Schools.
# Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER
# This file is distributed under the same license as the PACKAGE package.
#
msgid ""
msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2026-07-10 11:25+0200\n"
"PO-Revision-Date: 2025-12-04 13:00+0100\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: French <LL@li.org>\n"
"Language: fr\n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=UTF-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Plural-Forms: nplurals=2; plural=(n > 1);\n"
#: streetup/schools/templates/schools/base.html:7
msgid "Accueil"
msgstr "Accueil"
#: streetup/schools/templates/schools/base.html:11
#: streetup/schools/templates/schools/index.html:19
msgid "Géolocalisation"
msgstr "Géolocalisation"
#: streetup/schools/templates/schools/geolocation/index.html:11
msgid "Géolocalisation d'adresses"
msgstr "Géolocalisation d'adresses"
#: streetup/schools/templates/schools/geolocation/index.html:14
msgid ""
"Uploadez un fichier Excel contenant des adresses dans la première colonne "
"pour les géolocaliser automatiquement."
msgstr ""
"Uploadez un fichier Excel contenant des adresses dans la première colonne "
"pour les géolocaliser automatiquement."
#: streetup/schools/templates/schools/geolocation/index.html:36
msgid "Uploader un fichier"
msgstr "Uploader un fichier"
#: streetup/schools/templates/schools/geolocation/index.html:60
msgid "Format du fichier :"
msgstr "Format du fichier :"
#: streetup/schools/templates/schools/geolocation/index.html:62
msgid "Le fichier doit être au format Excel (.xls ou .xlsx)"
msgstr "Le fichier doit être au format Excel (.xls ou .xlsx)"
#: streetup/schools/templates/schools/geolocation/index.html:63
msgid "Les adresses doivent être dans la première colonne"
msgstr "Les adresses doivent être dans la première colonne"
#: streetup/schools/templates/schools/geolocation/index.html:64
msgid "Taille maximale : 10 Mo"
msgstr "Taille maximale : 10 Mo"
#: streetup/schools/templates/schools/geolocation/index.html:69
msgid "Lancer la géolocalisation"
msgstr "Lancer la géolocalisation"
#: streetup/schools/templates/schools/geolocation/index.html:77
msgid "Comment ça fonctionne ?"
msgstr "Comment ça fonctionne ?"
#: streetup/schools/templates/schools/geolocation/index.html:81
msgid "Préparez un fichier Excel avec les adresses dans la première colonne"
msgstr "Préparez un fichier Excel avec les adresses dans la première colonne"
#: streetup/schools/templates/schools/geolocation/index.html:82
msgid "Uploadez le fichier en utilisant le formulaire ci-dessus"
msgstr "Uploadez le fichier en utilisant le formulaire ci-dessus"
#: streetup/schools/templates/schools/geolocation/index.html:83
msgid "Le système géolocalise automatiquement chaque adresse"
msgstr "Le système géolocalise automatiquement chaque adresse"
#: streetup/schools/templates/schools/geolocation/index.html:84
msgid ""
"Vous recevrez un email avec le fichier résultat contenant les coordonnées "
"(longitude et latitude)"
msgstr ""
"Vous recevrez un email avec le fichier résultat contenant les coordonnées "
"(longitude et latitude)"
#: streetup/schools/templates/schools/geolocation/index.html:88
msgid ""
"Le traitement peut prendre plusieurs minutes selon le nombre d'adresses."
msgstr ""
"Le traitement peut prendre plusieurs minutes selon le nombre d'adresses."
#: streetup/schools/templates/schools/geolocation/index.html:97
msgid "Mes tâches de géolocalisation"
msgstr "Mes tâches de géolocalisation"
#: streetup/schools/templates/schools/geolocation/index.html:105
msgid "Fichier"
msgstr "Fichier"
#: streetup/schools/templates/schools/geolocation/index.html:106
msgid "Date"
msgstr "Date"
#: streetup/schools/templates/schools/geolocation/index.html:107
msgid "Statut"
msgstr "Statut"
#: streetup/schools/templates/schools/geolocation/index.html:108
msgid "Adresses"
msgstr "Adresses"
#: streetup/schools/templates/schools/geolocation/index.html:109
msgid "Réussies"
msgstr "Réussies"
#: streetup/schools/templates/schools/geolocation/index.html:110
msgid "Action"
msgstr "Action"
#: streetup/schools/templates/schools/geolocation/index.html:122
#: streetup/schools/templates/schools/geolocation/index.html:181
msgid "En attente"
msgstr "En attente"
#: streetup/schools/templates/schools/geolocation/index.html:124
#: streetup/schools/templates/schools/geolocation/index.html:182
msgid "En cours"
msgstr "En cours"
#: streetup/schools/templates/schools/geolocation/index.html:126
#: streetup/schools/templates/schools/geolocation/index.html:183
msgid "Terminé"
msgstr "Terminé"
#: streetup/schools/templates/schools/geolocation/index.html:128
#: streetup/schools/templates/schools/geolocation/index.html:184
msgid "Échoué"
msgstr "Échoué"
#: streetup/schools/templates/schools/geolocation/index.html:136
#: streetup/schools/templates/schools/geolocation/index.html:193
msgid "Télécharger"
msgstr "Télécharger"
#: streetup/schools/templates/schools/geolocation/index.html:140
#: streetup/schools/templates/schools/geolocation/index.html:197
msgid "Chargement..."
msgstr "Chargement..."
#: streetup/schools/templates/schools/geolocation/index.html:158
msgid "Aucune tâche de géolocalisation pour le moment."
msgstr "Aucune tâche de géolocalisation pour le moment."
#: streetup/schools/templates/schools/index.html:9
msgid "Écoles"
msgstr "Écoles"
#: streetup/schools/templates/schools/index.html:10
msgid "Bienvenue dans la gestion des écoles"
msgstr "Bienvenue dans la gestion des écoles"
#: streetup/schools/templates/schools/index.html:22
msgid ""
"Uploadez un fichier Excel contenant des adresses pour les géolocaliser "
"automatiquement."
msgstr ""
"Uploadez un fichier Excel contenant des adresses pour les géolocaliser "
"automatiquement."
#: streetup/schools/templates/schools/index.html:25
msgid "Accéder"
msgstr "Accéder"
#: streetup/schools/views.py:33 streetup/schools/views.py:51
msgid "Vous n'avez pas accès à cette vue."
msgstr "Vous n'avez pas accès à cette vue."
#: streetup/schools/views.py:73
msgid ""
"Votre fichier a été uploadé avec succès. La géolocalisation est en cours. "
"Vous recevrez un email lorsque le traitement sera terminé."
msgstr ""
"Votre fichier a été uploadé avec succès. La géolocalisation est en cours. "
"Vous recevrez un email lorsque le traitement sera terminé."
#: streetup/schools/views.py:268
msgid "Vous n'avez pas accès à ce fichier."
msgstr "Vous n'avez pas accès à ce fichier."
#: streetup/schools/views.py:272
msgid "Le fichier résultat n'est pas disponible."
msgstr "Le fichier résultat n'est pas disponible."
#: streetup/schools/views.py:276
msgid "Le fichier résultat n'a pas été trouvé."
msgstr "Le fichier résultat n'a pas été trouvé."

BIN
streetup/schools/locale/nl/LC_MESSAGES/django.mo
View file

Binary file not shown.

197
streetup/schools/locale/nl/LC_MESSAGES/django.po
View file

@ -1,197 +0,0 @@
# Nederlandse vertalingen voor de Schools module.
# Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER
# This file is distributed under the same license as the PACKAGE package.
#
msgid ""
msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2026-07-10 11:25+0200\n"
"PO-Revision-Date: 2025-12-04 13:00+0100\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: Dutch <LL@li.org>\n"
"Language: nl\n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=UTF-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Plural-Forms: nplurals=2; plural=(n != 1);\n"
#: streetup/schools/templates/schools/base.html:7
msgid "Accueil"
msgstr "Startpagina"
#: streetup/schools/templates/schools/base.html:11
#: streetup/schools/templates/schools/index.html:19
msgid "Géolocalisation"
msgstr "Geolocatie"
#: streetup/schools/templates/schools/geolocation/index.html:11
msgid "Géolocalisation d'adresses"
msgstr "Adresgeolocatie"
#: streetup/schools/templates/schools/geolocation/index.html:14
msgid ""
"Uploadez un fichier Excel contenant des adresses dans la première colonne "
"pour les géolocaliser automatiquement."
msgstr ""
"Upload een Excel-bestand met adressen in de eerste kolom om ze automatisch "
"te geolocaliseren."
#: streetup/schools/templates/schools/geolocation/index.html:36
msgid "Uploader un fichier"
msgstr "Bestand uploaden"
#: streetup/schools/templates/schools/geolocation/index.html:60
msgid "Format du fichier :"
msgstr "Bestandsformaat:"
#: streetup/schools/templates/schools/geolocation/index.html:62
msgid "Le fichier doit être au format Excel (.xls ou .xlsx)"
msgstr "Het bestand moet in Excel-formaat zijn (.xls of .xlsx)"
#: streetup/schools/templates/schools/geolocation/index.html:63
msgid "Les adresses doivent être dans la première colonne"
msgstr "De adressen moeten in de eerste kolom staan"
#: streetup/schools/templates/schools/geolocation/index.html:64
msgid "Taille maximale : 10 Mo"
msgstr "Maximale grootte: 10 MB"
#: streetup/schools/templates/schools/geolocation/index.html:69
msgid "Lancer la géolocalisation"
msgstr "Geolocatie starten"
#: streetup/schools/templates/schools/geolocation/index.html:77
msgid "Comment ça fonctionne ?"
msgstr "Hoe werkt het?"
#: streetup/schools/templates/schools/geolocation/index.html:81
msgid "Préparez un fichier Excel avec les adresses dans la première colonne"
msgstr "Bereid een Excel-bestand voor met adressen in de eerste kolom"
#: streetup/schools/templates/schools/geolocation/index.html:82
msgid "Uploadez le fichier en utilisant le formulaire ci-dessus"
msgstr "Upload het bestand via het bovenstaande formulier"
#: streetup/schools/templates/schools/geolocation/index.html:83
msgid "Le système géolocalise automatiquement chaque adresse"
msgstr "Het systeem geolocaliseert automatisch elk adres"
#: streetup/schools/templates/schools/geolocation/index.html:84
msgid ""
"Vous recevrez un email avec le fichier résultat contenant les coordonnées "
"(longitude et latitude)"
msgstr ""
"U ontvangt een e-mail met het resultaatbestand met de coördinaten "
"(lengtegraad en breedtegraad)"
#: streetup/schools/templates/schools/geolocation/index.html:88
msgid ""
"Le traitement peut prendre plusieurs minutes selon le nombre d'adresses."
msgstr ""
"De verwerking kan enkele minuten duren afhankelijk van het aantal adressen."
#: streetup/schools/templates/schools/geolocation/index.html:97
msgid "Mes tâches de géolocalisation"
msgstr "Mijn geolocatietaken"
#: streetup/schools/templates/schools/geolocation/index.html:105
msgid "Fichier"
msgstr "Bestand"
#: streetup/schools/templates/schools/geolocation/index.html:106
msgid "Date"
msgstr "Datum"
#: streetup/schools/templates/schools/geolocation/index.html:107
msgid "Statut"
msgstr "Status"
#: streetup/schools/templates/schools/geolocation/index.html:108
msgid "Adresses"
msgstr "Adressen"
#: streetup/schools/templates/schools/geolocation/index.html:109
msgid "Réussies"
msgstr "Geslaagd"
#: streetup/schools/templates/schools/geolocation/index.html:110
msgid "Action"
msgstr "Actie"
#: streetup/schools/templates/schools/geolocation/index.html:122
#: streetup/schools/templates/schools/geolocation/index.html:181
msgid "En attente"
msgstr "In afwachting"
#: streetup/schools/templates/schools/geolocation/index.html:124
#: streetup/schools/templates/schools/geolocation/index.html:182
msgid "En cours"
msgstr "In behandeling"
#: streetup/schools/templates/schools/geolocation/index.html:126
#: streetup/schools/templates/schools/geolocation/index.html:183
msgid "Terminé"
msgstr "Afgerond"
#: streetup/schools/templates/schools/geolocation/index.html:128
#: streetup/schools/templates/schools/geolocation/index.html:184
msgid "Échoué"
msgstr "Mislukt"
#: streetup/schools/templates/schools/geolocation/index.html:136
#: streetup/schools/templates/schools/geolocation/index.html:193
msgid "Télécharger"
msgstr "Downloaden"
#: streetup/schools/templates/schools/geolocation/index.html:140
#: streetup/schools/templates/schools/geolocation/index.html:197
msgid "Chargement..."
msgstr "Laden..."
#: streetup/schools/templates/schools/geolocation/index.html:158
msgid "Aucune tâche de géolocalisation pour le moment."
msgstr "Momenteel geen geolocatietaak."
#: streetup/schools/templates/schools/index.html:9
msgid "Écoles"
msgstr "Scholen"
#: streetup/schools/templates/schools/index.html:10
msgid "Bienvenue dans la gestion des écoles"
msgstr "Welkom bij het beheer van scholen"
#: streetup/schools/templates/schools/index.html:22
msgid ""
"Uploadez un fichier Excel contenant des adresses pour les géolocaliser "
"automatiquement."
msgstr ""
"Upload een Excel-bestand met adressen om ze automatisch te geolocaliseren."
#: streetup/schools/templates/schools/index.html:25
msgid "Accéder"
msgstr "Toegang"
#: streetup/schools/views.py:33 streetup/schools/views.py:51
msgid "Vous n'avez pas accès à cette vue."
msgstr "U hebt geen toegang tot deze weergave."
#: streetup/schools/views.py:73
msgid ""
"Votre fichier a été uploadé avec succès. La géolocalisation est en cours. "
"Vous recevrez un email lorsque le traitement sera terminé."
msgstr ""
"Uw bestand is succesvol geüpload. De geolocatie is bezig. U ontvangt een e-"
"mail wanneer de verwerking is voltooid."
#: streetup/schools/views.py:268
msgid "Vous n'avez pas accès à ce fichier."
msgstr "U hebt geen toegang tot dit bestand."
#: streetup/schools/views.py:272
msgid "Le fichier résultat n'est pas disponible."
msgstr "Het resultaatbestand is niet beschikbaar."
#: streetup/schools/views.py:276
msgid "Le fichier résultat n'a pas été trouvé."
msgstr "Het resultaatbestand is niet gevonden."

0
streetup/schools/management/__init__.py
View file

0
streetup/schools/management/commands/__init__.py
View file

40
streetup/schools/management/commands/setup_schools_view.py
View file

@ -1,40 +0,0 @@
"""
Management command to set up the Schools AppView.
Usage:
python manage.py setup_schools_view
"""
from django.core.management.base import BaseCommand
from common.models import AppView
class Command(BaseCommand):
help = 'Creates the Schools AppView entry for navigation and permissions'
def handle(self, *args, **options):
schools_view, created = AppView.objects.get_or_create(
code='schools',
defaults={
'name_fr': 'Écoles',
'name_nl': 'Scholen',
'icon': 'bi-building',
'url_name': 'schools:index',
'is_default': False,
'order': 30
}
)
if created:
self.stdout.write(
self.style.SUCCESS(f'✓ Successfully created Schools AppView (ID: {schools_view.id})')
)
else:
self.stdout.write(
self.style.WARNING(f'⚠ Schools AppView already exists (ID: {schools_view.id})')
)
self.stdout.write('\nNext steps:')
self.stdout.write('1. Grant access to users via Django admin or shell:')
self.stdout.write(' user.config.accessible_views.add(schools_view)')
self.stdout.write('2. Configure email settings in .env or settings.py')
self.stdout.write('3. Create media directories for file uploads')

40
streetup/schools/migrations/0001_initial.py
View file

@ -1,40 +0,0 @@
# Generated by Django 6.0.7 on 2026-07-22 13:42
import common.storages
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='GeocodingTask',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('uploaded_file', models.FileField(storage=common.storages.PrivateMediaStorage(), upload_to='schools/geocoding/uploads/%Y/%m/%d/', verbose_name='Fichier uploadé')),
('output_file', models.FileField(blank=True, null=True, storage=common.storages.PrivateMediaStorage(), upload_to='schools/geocoding/outputs/%Y/%m/%d/', verbose_name='Fichier résultat')),
('status', models.CharField(choices=[('pending', 'En attente'), ('processing', 'En cours'), ('completed', 'Terminé'), ('failed', 'Échoué')], default='pending', max_length=20, verbose_name='Statut')),
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='Créé le')),
('started_at', models.DateTimeField(blank=True, null=True, verbose_name='Démarré le')),
('completed_at', models.DateTimeField(blank=True, null=True, verbose_name='Terminé le')),
('error_message', models.TextField(blank=True, null=True, verbose_name="Message d'erreur")),
('total_rows', models.IntegerField(default=0, verbose_name='Nombre total de lignes')),
('successful_geocodes', models.IntegerField(default=0, verbose_name='Géolocalisations réussies')),
('failed_geocodes', models.IntegerField(default=0, verbose_name='Géolocalisations échouées')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL, verbose_name='Utilisateur')),
],
options={
'verbose_name': 'Tâche de géolocalisation',
'verbose_name_plural': 'Tâches de géolocalisation',
'ordering': ['-created_at'],
},
),
]

0
streetup/schools/migrations/__init__.py
View file

53
streetup/schools/models.py
View file

@ -1,53 +0,0 @@
from django.db import models
from django.contrib.auth.models import User
from django.utils import timezone
from common.storages import PrivateMediaStorage
private_storage = PrivateMediaStorage()
class GeocodingTask(models.Model):
"""Model to track geocoding tasks"""
STATUS_CHOICES = [
('pending', 'En attente'),
('processing', 'En cours'),
('completed', 'Terminé'),
('failed', 'Échoué'),
]
user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name="Utilisateur")
uploaded_file = models.FileField(
upload_to='schools/geocoding/uploads/%Y/%m/%d/',
storage=private_storage,
verbose_name="Fichier uploadé"
)
output_file = models.FileField(
upload_to='schools/geocoding/outputs/%Y/%m/%d/',
storage=private_storage,
verbose_name="Fichier résultat",
blank=True,
null=True
)
status = models.CharField(
max_length=20,
choices=STATUS_CHOICES,
default='pending',
verbose_name="Statut"
)
created_at = models.DateTimeField(auto_now_add=True, verbose_name="Créé le")
started_at = models.DateTimeField(null=True, blank=True, verbose_name="Démarré le")
completed_at = models.DateTimeField(null=True, blank=True, verbose_name="Terminé le")
error_message = models.TextField(blank=True, null=True, verbose_name="Message d'erreur")
# Statistics
total_rows = models.IntegerField(default=0, verbose_name="Nombre total de lignes")
successful_geocodes = models.IntegerField(default=0, verbose_name="Géolocalisations réussies")
failed_geocodes = models.IntegerField(default=0, verbose_name="Géolocalisations échouées")
class Meta:
verbose_name = "Tâche de géolocalisation"
verbose_name_plural = "Tâches de géolocalisation"
ordering = ['-created_at']
def __str__(self):
return f"Tâche {self.id} - {self.user.username} - {self.status}"

13
streetup/schools/templates/schools/base.html
View file

@ -1,13 +0,0 @@
{% extends "base.html" %}
{% load i18n %}
{% block sidebar %}
<a href="{% url 'schools:index' %}" class="nav-link d-flex align-items-center p-2 {% if request.path == '/schools/' %}active{% endif %}">
<i class="bi bi-house-fill" data-bs-toggle="tooltip" data-bs-placement="right" title="Accueil"></i>
<span class="ms-2 text d-none d-md-inline">{% translate "Accueil" %}</span>
</a>
<a href="{% url 'schools:geolocation' %}" class="nav-link d-flex align-items-center p-2 {% if request.path|slice:':20' == '/schools/geolocation' %}active{% endif %}">
<i class="bi bi-geo-alt" data-bs-toggle="tooltip" data-bs-placement="right" title="Géolocalisation"></i>
<span class="ms-2 text d-none d-md-inline">{% translate "Géolocalisation" %}</span>
</a>
{% endblock sidebar %}

293
streetup/schools/templates/schools/geolocation/index.html
View file

@ -1,293 +0,0 @@
{% extends "schools/base.html" %}
{% load i18n %}
{% load static %}
{% load short_name %}
{% block content %}
<div class="container-fluid mt-4">
<div class="row">
<div class="col-12">
<h1>
<i class="bi bi-geo-alt me-2"></i>{% translate "Géolocalisation d'adresses" %}
</h1>
<p class="lead">
{% translate "Uploadez un fichier Excel contenant des adresses dans la première colonne pour les géolocaliser automatiquement." %}
</p>
</div>
</div>
{% if messages %}
<div class="row mt-3">
<div class="col-12">
{% for message in messages %}
<div class="alert alert-{{ message.tags }} alert-dismissible fade show" role="alert">
{{ message }}
<button type="button" class="btn-close" data-bs-dismiss="alert" aria-label="Close"></button>
</div>
{% endfor %}
</div>
</div>
{% endif %}
<div class="row mt-4">
<div class="col-lg-6">
<div class="card">
<div class="card-header bg-primary text-white">
<h5 class="mb-0">{% translate "Uploader un fichier" %}</h5>
</div>
<div class="card-body">
<form method="post" enctype="multipart/form-data">
{% csrf_token %}
<div class="mb-3">
<label for="{{ form.uploaded_file.id_for_label }}" class="form-label">
{{ form.uploaded_file.label }}
</label>
{{ form.uploaded_file }}
{% if form.uploaded_file.help_text %}
<div class="form-text">{{ form.uploaded_file.help_text }}</div>
{% endif %}
{% if form.uploaded_file.errors %}
<div class="text-danger">
{% for error in form.uploaded_file.errors %}
{{ error }}
{% endfor %}
</div>
{% endif %}
</div>
<div class="alert alert-info" role="alert">
<strong>{% translate "Format du fichier :" %}</strong>
<ul class="mb-0 mt-2">
<li>{% translate "Le fichier doit être au format Excel (.xls ou .xlsx)" %}</li>
<li>{% translate "Les adresses doivent être dans la première colonne" %}</li>
<li>{% translate "Taille maximale : 10 Mo" %}</li>
</ul>
</div>
<button type="submit" class="btn btn-primary">
<i class="bi bi-upload me-2"></i>{% translate "Lancer la géolocalisation" %}
</button>
</form>
</div>
</div>
<div class="card mt-4">
<div class="card-header">
<h5 class="mb-0">{% translate "Comment ça fonctionne ?" %}</h5>
</div>
<div class="card-body">
<ol>
<li>{% translate "Préparez un fichier Excel avec les adresses dans la première colonne" %}</li>
<li>{% translate "Uploadez le fichier en utilisant le formulaire ci-dessus" %}</li>
<li>{% translate "Le système géolocalise automatiquement chaque adresse" %}</li>
<li>{% translate "Vous recevrez un email avec le fichier résultat contenant les coordonnées (longitude et latitude)" %}</li>
</ol>
<div class="alert alert-warning mt-3" role="alert">
<i class="bi bi-exclamation-triangle me-2"></i>
{% translate "Le traitement peut prendre plusieurs minutes selon le nombre d'adresses." %}
</div>
</div>
</div>
</div>
<div class="col-lg-6">
<div class="card">
<div class="card-header">
<h5 class="mb-0">{% translate "Mes tâches de géolocalisation" %}</h5>
</div>
<div class="card-body">
{% if user_tasks %}
<div class="table-responsive">
<table class="table table-hover">
<thead>
<tr>
<th>{% translate "Fichier" %}</th>
<th>{% translate "Date" %}</th>
<th>{% translate "Statut" %}</th>
<th>{% translate "Adresses" %}</th>
<th>{% translate "Réussies" %}</th>
<th>{% translate "Action" %}</th>
</tr>
</thead>
<tbody>
{% for task in user_tasks %}
<tr>
<td class="text-truncate" style="max-width: 200px;" title="{{ task.uploaded_file|filename }}">
{{ task.uploaded_file|filename|default:"-" }}
</td>
<td>{{ task.created_at|date:"d/m/Y H:i" }}</td>
<td>
{% if task.status == 'pending' %}
<span class="badge bg-secondary">{% translate "En attente" %}</span>
{% elif task.status == 'processing' %}
<span class="badge bg-info">{% translate "En cours" %}</span>
{% elif task.status == 'completed' %}
<span class="badge bg-success">{% translate "Terminé" %}</span>
{% elif task.status == 'failed' %}
<span class="badge bg-danger">{% translate "Échoué" %}</span>
{% endif %}
</td>
<td>{{ task.total_rows|default:"-" }}</td>
<td>{{ task.successful_geocodes|default:"-" }}</td>
<td>
{% if task.status == 'completed' and task.output_file %}
<a href="{% url 'schools:download_result' task.id %}" class="btn btn-sm btn-primary" download>
<i class="bi bi-download"></i> {% translate "Télécharger" %}
</a>
{% elif task.status == 'processing' %}
<span class="spinner-border spinner-border-sm" role="status">
<span class="visually-hidden">{% translate "Chargement..." %}</span>
</span>
{% elif task.status == 'failed' %}
<button type="button" class="btn btn-sm btn-outline-danger"
data-bs-toggle="tooltip"
title="{{ task.error_message|default:'Erreur inconnue' }}">
<i class="bi bi-exclamation-circle"></i>
</button>
{% else %}
-
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% else %}
<p class="text-muted">{% translate "Aucune tâche de géolocalisation pour le moment." %}</p>
{% endif %}
</div>
</div>
</div>
</div>
</div>
{% endblock content %}
{% block script %}
<script>
// Enable tooltips
var tooltipTriggerList = [].slice.call(document.querySelectorAll('[data-bs-toggle="tooltip"]'))
var tooltipList = tooltipTriggerList.map(function (tooltipTriggerEl) {
return new bootstrap.Tooltip(tooltipTriggerEl)
});
// Auto-refresh task statuses every 15 seconds
const REFRESH_INTERVAL = 15000; // 15 seconds
let refreshTimer = null;
function getStatusBadge(status) {
const statusMap = {
'pending': { class: 'bg-secondary', text: '{% translate "En attente" %}' },
'processing': { class: 'bg-info', text: '{% translate "En cours" %}' },
'completed': { class: 'bg-success', text: '{% translate "Terminé" %}' },
'failed': { class: 'bg-danger', text: '{% translate "Échoué" %}' }
};
const info = statusMap[status] || { class: 'bg-secondary', text: status };
return `<span class="badge ${info.class}">${info.text}</span>`;
}
function getActionButton(task) {
if (task.status === 'completed' && task.download_url) {
return `<a href="${task.download_url}" class="btn btn-sm btn-primary" download>
<i class="bi bi-download"></i> {% translate "Télécharger" %}
</a>`;
} else if (task.status === 'processing') {
return `<span class="spinner-border spinner-border-sm" role="status">
<span class="visually-hidden">{% translate "Chargement..." %}</span>
</span>`;
} else if (task.status === 'failed') {
return `<button type="button" class="btn btn-sm btn-outline-danger"
data-bs-toggle="tooltip"
title="${task.error_message}">
<i class="bi bi-exclamation-circle"></i>
</button>`;
}
return '-';
}
function updateTasksTable(tasks) {
const tbody = document.querySelector('.table tbody');
if (!tbody) return;
// Clear existing rows
tbody.innerHTML = '';
tasks.forEach(task => {
const row = document.createElement('tr');
row.innerHTML = `
<td class="text-truncate" style="max-width: 200px;" title="${task.filename}">
${task.filename}
</td>
<td>${task.created_at}</td>
<td>${getStatusBadge(task.status)}</td>
<td>${task.total_rows}</td>
<td>${task.successful_geocodes}</td>
<td>${getActionButton(task)}</td>
`;
tbody.appendChild(row);
});
// Re-initialize tooltips for new elements
var newTooltipTriggerList = [].slice.call(tbody.querySelectorAll('[data-bs-toggle="tooltip"]'));
newTooltipTriggerList.forEach(function (tooltipTriggerEl) {
new bootstrap.Tooltip(tooltipTriggerEl);
});
}
function refreshTasksStatus() {
fetch('{% url "schools:geolocation_tasks_status" %}')
.then(response => response.json())
.then(data => {
if (data.tasks && data.tasks.length > 0) {
updateTasksTable(data.tasks);
}
// Continue refreshing only if there are active tasks
if (data.has_active_tasks) {
scheduleRefresh();
} else {
stopRefresh();
}
})
.catch(error => {
console.error('Error fetching task status:', error);
// Continue trying even on error
scheduleRefresh();
});
}
function scheduleRefresh() {
if (!refreshTimer) {
refreshTimer = setTimeout(() => {
refreshTimer = null;
refreshTasksStatus();
}, REFRESH_INTERVAL);
}
}
function stopRefresh() {
if (refreshTimer) {
clearTimeout(refreshTimer);
refreshTimer = null;
}
}
// Start auto-refresh if there are pending or processing tasks
document.addEventListener('DOMContentLoaded', function() {
{% if user_tasks %}
const hasActiveTasks = {% if user_tasks.0.status == 'pending' or user_tasks.0.status == 'processing' %}true{% else %}false{% endif %};
// Check all tasks for active status
const activeStatuses = ['pending', 'processing'];
{% for task in user_tasks %}
{% if task.status == 'pending' or task.status == 'processing' %}
scheduleRefresh();
{% endif %}
{% endfor %}
{% endif %}
});
// Stop refresh when leaving the page
window.addEventListener('beforeunload', stopRefresh);
</script>
{% endblock script %}

32
streetup/schools/templates/schools/index.html
View file

@ -1,32 +0,0 @@
{% extends "schools/base.html" %}
{% load i18n %}
{% load static %}
{% block content %}
<div class="container-fluid mt-4">
<div class="row">
<div class="col-12">
<h1>{% translate "Écoles" %}</h1>
<p class="lead">{% translate "Bienvenue dans la gestion des écoles" %}</p>
</div>
</div>
<div class="row mt-4">
<div class="col-md-6 col-lg-4">
<div class="card">
<div class="card-body">
<h5 class="card-title">
<i class="bi bi-geo-alt me-2"></i>{% translate "Géolocalisation" %}
</h5>
<p class="card-text">
{% translate "Uploadez un fichier Excel contenant des adresses pour les géolocaliser automatiquement." %}
</p>
<a href="{% url 'schools:geolocation' %}" class="btn btn-primary">
{% translate "Accéder" %}
</a>
</div>
</div>
</div>
</div>
</div>
{% endblock content %}

230
streetup/schools/test_internal_geocoding.py
View file

@ -1,230 +0,0 @@
from django.test import TestCase
from common.models import Municipality, AdrStreet, AdrPoint
from django.contrib.gis.geos import Point
from .internal_geocoding import (
normalize_address,
search_address_points,
search_streets,
geocode_internal,
DEFAULT_SCORE_THRESHOLD
)
class NormalizeAddressTestCase(TestCase):
"""Test address normalization"""
def test_normalize_empty_address(self):
self.assertEqual(normalize_address(""), "")
self.assertEqual(normalize_address(None), "")
def test_normalize_removes_extra_spaces(self):
self.assertEqual(normalize_address(" rue de la loi "), "rue de la loi")
def test_normalize_lowercase(self):
self.assertEqual(normalize_address("Rue DE La LOI"), "rue de la loi")
class InternalGeocodingTestCase(TestCase):
"""Test internal geocoding with fuzzy matching"""
def setUp(self):
"""Create test data"""
# Create a municipality
self.municipality = Municipality.objects.create(
fid="1000",
name_fr="Bruxelles",
name_nl="Brussel",
niscode="21004"
)
# Create a street
self.street = AdrStreet.objects.create(
fid="street_001",
name_fr="Rue de la Loi",
name_nl="Wetstraat",
full_street_fr="Rue de la Loi",
full_street_nl="Wetstraat",
municipality=self.municipality,
lon=4.3676,
lat=50.8467,
centroid=Point(4.3676, 50.8467, srid=3812)
)
# Create address points
self.point1 = AdrPoint.objects.create(
fid="point_001",
number="16",
street_name=self.street,
municipality=self.municipality,
zipcode="1000",
full_address_fr="16 Rue de la Loi Bruxelles",
full_address_nl="16 Wetstraat Brussel",
lon=4.3676,
lat=50.8467,
geom=Point(4.3676, 50.8467, srid=3812)
)
self.point2 = AdrPoint.objects.create(
fid="point_002",
number="100",
street_name=self.street,
municipality=self.municipality,
zipcode="1000",
full_address_fr="100 Rue de la Loi Bruxelles",
full_address_nl="100 Wetstraat Brussel",
lon=4.3680,
lat=50.8470,
geom=Point(4.3680, 50.8470, srid=3812)
)
def test_search_address_points_exact_match(self):
"""Test searching for an exact address match"""
results = search_address_points("16 Rue de la Loi Bruxelles", limit=5)
self.assertGreater(len(results), 0)
point, score = results[0]
self.assertEqual(point.fid, "point_001")
self.assertGreater(score, 90) # Should be a very high score
def test_search_address_points_french(self):
"""Test searching in French"""
results = search_address_points("16 rue de la loi", limit=5)
self.assertGreater(len(results), 0)
point, score = results[0]
self.assertEqual(point.number, "16")
self.assertGreater(score, 60)
def test_search_address_points_dutch(self):
"""Test searching in Dutch"""
results = search_address_points("16 wetstraat", limit=5)
self.assertGreater(len(results), 0)
point, score = results[0]
self.assertEqual(point.number, "16")
self.assertGreater(score, 60)
def test_search_streets_french(self):
"""Test searching for streets in French"""
results = search_streets("Rue de la Loi", limit=5)
self.assertGreater(len(results), 0)
street, score = results[0]
self.assertEqual(street.fid, "street_001")
self.assertGreater(score, 90)
def test_search_streets_dutch(self):
"""Test searching for streets in Dutch"""
results = search_streets("Wetstraat", limit=5)
self.assertGreater(len(results), 0)
street, score = results[0]
self.assertEqual(street.fid, "street_001")
self.assertGreater(score, 90)
def test_geocode_internal_success(self):
"""Test successful internal geocoding"""
lon, lat, found_addr, score = geocode_internal("16 Rue de la Loi", threshold=60)
self.assertIsNotNone(lon)
self.assertIsNotNone(lat)
self.assertIn("16", found_addr)
self.assertIn("Rue de la Loi", found_addr)
self.assertGreater(score, 60)
def test_geocode_internal_street_only(self):
"""Test geocoding with street name only (no house number)"""
lon, lat, found_addr, score = geocode_internal("Rue de la Loi Bruxelles", threshold=60)
self.assertIsNotNone(lon)
self.assertIsNotNone(lat)
self.assertIn("Rue de la Loi", found_addr)
self.assertGreater(score, 60)
def test_geocode_internal_below_threshold(self):
"""Test that low-scoring matches are rejected"""
# Use a very high threshold that won't be met
lon, lat, found_addr, score = geocode_internal("16 Rue de la Loi", threshold=99)
# Should return None if score is below threshold
# (assuming the score is less than 99)
if score < 99:
self.assertIsNone(lon)
self.assertIsNone(lat)
def test_geocode_internal_no_match(self):
"""Test geocoding with no matching address"""
lon, lat, found_addr, score = geocode_internal("Nonexistent Street 999", threshold=80)
self.assertIsNone(lon)
self.assertIsNone(lat)
self.assertEqual(score, 0)
def test_geocode_internal_empty_address(self):
"""Test geocoding with empty address"""
lon, lat, found_addr, score = geocode_internal("", threshold=80)
self.assertIsNone(lon)
self.assertIsNone(lat)
self.assertEqual(score, 0)
def test_geocode_internal_fuzzy_match(self):
"""Test fuzzy matching with typos"""
# Test with a typo: "Rua" instead of "Rue"
lon, lat, found_addr, score = geocode_internal("16 Rua de la Loi", threshold=70)
# Should still find a match with a reasonable score
if score >= 70:
self.assertIsNotNone(lon)
self.assertIsNotNone(lat)
def test_geocode_internal_partial_match(self):
"""Test partial address matching"""
lon, lat, found_addr, score = geocode_internal("loi 16", threshold=70)
# Should find a match even with different word order
if score >= 70:
self.assertIsNotNone(lon)
self.assertIsNotNone(lat)
class SearchLimitTestCase(TestCase):
"""Test search limit functionality"""
def setUp(self):
"""Create multiple test addresses"""
self.municipality = Municipality.objects.create(
fid="1000",
name_fr="Bruxelles",
name_nl="Brussel",
niscode="21004"
)
self.street = AdrStreet.objects.create(
fid="street_001",
name_fr="Rue de la Loi",
name_nl="Wetstraat",
full_street_fr="Rue de la Loi",
full_street_nl="Wetstraat",
municipality=self.municipality,
lon=4.3676,
lat=50.8467,
centroid=Point(4.3676, 50.8467, srid=3812)
)
# Create 5 address points
for i in range(1, 6):
number = str(i * 10)
AdrPoint.objects.create(
fid=f"point_00{i}",
number=number,
street_name=self.street,
municipality=self.municipality,
zipcode="1000",
full_address_fr=f"{number} Rue de la Loi Bruxelles",
full_address_nl=f"{number} Wetstraat Brussel",
lon=4.3676 + i * 0.001,
lat=50.8467 + i * 0.001,
geom=Point(4.3676 + i * 0.001, 50.8467 + i * 0.001, srid=3812)
)
def test_search_respects_limit(self):
"""Test that search limit is respected"""
results = search_address_points("Rue de la Loi", limit=3)
self.assertLessEqual(len(results), 3)
def test_search_returns_best_matches(self):
"""Test that search returns best matches first"""
results = search_address_points("Rue de la Loi 10", limit=5)
if len(results) > 1:
# Scores should be in descending order
scores = [score for _, score in results]
self.assertEqual(scores, sorted(scores, reverse=True))

187
streetup/schools/tests.py
View file

@ -1,187 +0,0 @@
from django.test import TestCase, Client
from django.contrib.auth.models import User
from django.core.files.uploadedfile import SimpleUploadedFile
from django.urls import reverse
from common.models import UserConfig, AppView
from .models import GeocodingTask
from .forms import GeocodingUploadForm
import io
class SchoolsAppViewTestCase(TestCase):
"""Test cases for the Schools app view setup"""
def setUp(self):
# Create a test user
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123'
)
# Create user config
self.user_config = UserConfig.objects.create(user=self.user)
# Create Schools AppView
self.schools_view = AppView.objects.create(
code='schools',
name_fr='Écoles',
name_nl='Scholen',
icon='bi-building',
url_name='schools:index',
is_default=False,
order=30
)
def test_appview_created(self):
"""Test that Schools AppView can be created"""
self.assertEqual(self.schools_view.code, 'schools')
self.assertEqual(self.schools_view.name_fr, 'Écoles')
self.assertEqual(self.schools_view.url_name, 'schools:index')
def test_user_without_permission(self):
"""Test that user without permission cannot access schools view"""
client = Client()
client.login(username='testuser', password='testpass123')
response = client.get(reverse('schools:index'))
self.assertEqual(response.status_code, 403) # Permission denied
def test_user_with_permission(self):
"""Test that user with permission can access schools view"""
# Grant access to schools view
self.user_config.accessible_views.add(self.schools_view)
client = Client()
client.login(username='testuser', password='testpass123')
response = client.get(reverse('schools:index'))
self.assertEqual(response.status_code, 200)
class GeocodingTaskModelTestCase(TestCase):
"""Test cases for the GeocodingTask model"""
def setUp(self):
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123'
)
def test_geocoding_task_creation(self):
"""Test that GeocodingTask can be created"""
task = GeocodingTask.objects.create(
user=self.user,
status='pending'
)
self.assertEqual(task.user, self.user)
self.assertEqual(task.status, 'pending')
self.assertEqual(task.total_rows, 0)
def test_geocoding_task_str(self):
"""Test string representation of GeocodingTask"""
task = GeocodingTask.objects.create(
user=self.user,
status='completed'
)
expected = f"Tâche {task.id} - {self.user.username} - completed"
self.assertEqual(str(task), expected)
class GeocodingUploadFormTestCase(TestCase):
"""Test cases for the GeocodingUploadForm"""
def test_valid_excel_file(self):
"""Test form with valid Excel file"""
# Create a simple Excel file in memory
excel_content = b'PK\x03\x04' # Minimal xlsx signature
file = SimpleUploadedFile(
"test.xlsx",
excel_content,
content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
)
form = GeocodingUploadForm(files={'uploaded_file': file})
# Note: The form may be valid or invalid depending on actual file content
# This test just ensures the form can handle file input
self.assertIn('uploaded_file', form.fields)
def test_invalid_file_extension(self):
"""Test form rejects non-Excel files"""
file = SimpleUploadedFile(
"test.txt",
b"test content",
content_type="text/plain"
)
form = GeocodingUploadForm(files={'uploaded_file': file})
form.is_valid()
self.assertIn('uploaded_file', form.errors)
def test_file_too_large(self):
"""Test form rejects files larger than 10MB"""
# Create a file larger than 10MB
large_content = b'x' * (11 * 1024 * 1024) # 11MB
file = SimpleUploadedFile(
"large.xlsx",
large_content,
content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
)
form = GeocodingUploadForm(files={'uploaded_file': file})
form.is_valid()
self.assertIn('uploaded_file', form.errors)
class GeolocationViewTestCase(TestCase):
"""Test cases for the geolocation views"""
def setUp(self):
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123'
)
self.user_config = UserConfig.objects.create(user=self.user)
from django.utils import translation
translation.activate('fr')
self.schools_view = AppView.objects.create(
code='schools',
name_fr='Écoles',
name_nl='Scholen',
icon='bi-building',
url_name='schools:index',
is_default=False,
order=30
)
# Grant access
self.user_config.accessible_views.add(self.schools_view)
self.client = Client()
self.client.login(username='testuser', password='testpass123')
def test_geolocation_view_loads(self):
"""Test that geolocation view loads successfully"""
response = self.client.get(reverse('schools:geolocation'))
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'Géolocalisation')
def test_geolocation_view_shows_user_tasks(self):
"""Test that user's tasks are displayed"""
# Create a task for the user
task = GeocodingTask.objects.create(
user=self.user,
status='completed',
total_rows=10,
successful_geocodes=8
)
response = self.client.get(reverse('schools:geolocation'))
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'Completed')

11
streetup/schools/urls.py
View file

@ -1,11 +0,0 @@
from django.urls import path
from . import views
app_name = "schools"
urlpatterns = [
path("", views.index, name="index"),
path("geolocation/", views.geolocation, name="geolocation"),
path("geolocation/download/<int:task_id>/", views.download_result, name="download_result"),
path("geolocation/tasks-status/", views.geolocation_tasks_status, name="geolocation_tasks_status"),
]

329
streetup/schools/views.py
View file

@ -1,329 +0,0 @@
from django.shortcuts import render, redirect, get_object_or_404
from django.contrib.auth.decorators import login_required
from django.contrib import messages
from django.core.exceptions import PermissionDenied
from django.utils.translation import gettext as _
from django.utils import timezone
from django.conf import settings
from django.core.mail import send_mail
from django.urls import reverse
from django.http import FileResponse, Http404, JsonResponse
import threading
import os
import logging
from common.models import UserConfig
from .models import GeocodingTask
from .forms import GeocodingUploadForm
from .geocoding import geocode_excel_file
logger = logging.getLogger(__name__)
@login_required
def index(request):
"""
Schools view - main landing page
"""
user_config = get_object_or_404(UserConfig, user=request.user)
# Check if user has access to schools view
if not user_config.can_access_view('schools'):
raise PermissionDenied(_("Vous n'avez pas accès à cette vue."))
context = {
'user_config': user_config,
}
return render(request, "schools/index.html", context)
@login_required
def geolocation(request):
"""
Geolocation view - Upload and process Excel files for geocoding
"""
user_config = get_object_or_404(UserConfig, user=request.user)
# Check if user has access to schools view
if not user_config.can_access_view('schools'):
raise PermissionDenied(_("Vous n'avez pas accès à cette vue."))
if request.method == 'POST':
form = GeocodingUploadForm(request.POST, request.FILES)
if form.is_valid():
# Create the task
task = form.save(commit=False)
task.user = request.user
task.status = 'pending'
task.save()
# Start geocoding in background thread
# Note: Using non-daemon thread to prevent data loss on server shutdown
thread = threading.Thread(
target=process_geocoding_task,
args=(task.id,),
daemon=False
)
thread.start()
messages.success(
request,
_("Votre fichier a été uploadé avec succès. "
"La géolocalisation est en cours. "
"Vous recevrez un email lorsque le traitement sera terminé.")
)
return redirect('schools:geolocation')
else:
form = GeocodingUploadForm()
# Get user's tasks
user_tasks = GeocodingTask.objects.filter(user=request.user).order_by('-created_at')[:10]
context = {
'user_config': user_config,
'form': form,
'user_tasks': user_tasks,
}
return render(request, "schools/geolocation/index.html", context)
def process_geocoding_task(task_id):
"""
Background task to process geocoding
This runs in a separate thread
"""
task = None
try:
task = GeocodingTask.objects.get(id=task_id)
# Update status to processing
task.status = 'processing'
task.started_at = timezone.now()
task.save()
# Get file paths - use PRIVATE_MEDIA_ROOT for secure storage
input_path = task.uploaded_file.path
output_filename = f"geocoded_{os.path.basename(task.uploaded_file.name)}"
output_relative_path = os.path.join(
'schools/geocoding/outputs',
timezone.now().strftime('%Y/%m/%d'),
output_filename
)
output_path = os.path.join(
settings.PRIVATE_MEDIA_ROOT,
output_relative_path
)
# Ensure output directory exists
os.makedirs(os.path.dirname(output_path), exist_ok=True)
# Run geocoding
logger.info(f"Starting geocoding task {task_id}")
chunk_size = getattr(settings, 'SCHOOLS_GEOCODING_CHUNK_SIZE', 3)
api_limit = getattr(settings, 'SCHOOLS_GEOCODING_API_LIMIT', 1)
internal_threshold = getattr(settings, 'SCHOOLS_GEOCODING_INTERNAL_THRESHOLD', 80)
use_internal = getattr(settings, 'SCHOOLS_GEOCODING_USE_INTERNAL', True)
stats = geocode_excel_file(
input_path=input_path,
output_path=output_path,
chunk_size=chunk_size,
limit=api_limit,
output_sheet="geocoded",
internal_threshold=internal_threshold,
use_internal=use_internal
)
# Update task with results
task.status = 'completed'
task.completed_at = timezone.now()
task.total_rows = stats['total']
task.successful_geocodes = stats['success']
task.failed_geocodes = stats['failed']
# Save the output file path relative to PRIVATE_MEDIA_ROOT
task.output_file.name = output_relative_path
task.save()
logger.info(f"Completed geocoding task {task_id}")
# Send email notification
send_completion_email(task)
except Exception as e:
logger.error(f"Error processing geocoding task {task_id}: {str(e)}")
try:
# Retrieve task if not already loaded
if task is None:
task = GeocodingTask.objects.get(id=task_id)
task.status = 'failed'
task.completed_at = timezone.now()
task.error_message = str(e)
task.save()
# Send error email
send_error_email(task)
except Exception as inner_e:
logger.error(f"Error updating task status for task {task_id}: {str(inner_e)}")
def send_completion_email(task):
"""Send email notification when geocoding is complete"""
try:
user_email = task.user.email
if not user_email:
logger.warning(f"No email address for user {task.user.username}")
return
subject = "Géolocalisation terminée - StreetUp"
# Build download URL
# Use ROOT_URL from settings
root_url = getattr(settings, 'ROOT_URL', 'http://localhost:8000')
download_url = f"{root_url.rstrip('/')}{reverse('schools:download_result', args=[task.id])}"
message = f"""
Bonjour {task.user.first_name or task.user.username},
Votre tâche de géolocalisation est terminée.
Statistiques :
- Total d'adresses : {task.total_rows}
- Géolocalisations réussies : {task.successful_geocodes}
- Échecs : {task.failed_geocodes}
Vous pouvez télécharger le fichier résultat ici :
{download_url}
Cordialement,
L'équipe StreetUp
"""
try:
send_mail(
subject=subject,
message=message,
from_email=getattr(settings, 'DEFAULT_FROM_EMAIL', 'noreply@streetup.be'),
recipient_list=[user_email],
fail_silently=False,
)
logger.info(f"Sent completion email to {user_email} for task {task.id}")
except Exception as mail_error:
logger.error(f"Failed to send completion email to {user_email} for task {task.id}: {str(mail_error)}")
except Exception as e:
logger.error(f"Error sending completion email for task {task.id}: {str(e)}")
def send_error_email(task):
"""Send email notification when geocoding fails"""
try:
user_email = task.user.email
if not user_email:
logger.warning(f"No email address for user {task.user.username}")
return
subject = "Erreur lors de la géolocalisation - StreetUp"
message = f"""
Bonjour {task.user.first_name or task.user.username},
Une erreur est survenue lors du traitement de votre tâche de géolocalisation.
Détails de l'erreur :
{task.error_message}
Veuillez réessayer ou contacter l'administrateur si le problème persiste.
Cordialement,
L'équipe StreetUp
"""
try:
send_mail(
subject=subject,
message=message,
from_email=getattr(settings, 'DEFAULT_FROM_EMAIL', 'noreply@streetup.be'),
recipient_list=[user_email],
fail_silently=False,
)
logger.info(f"Sent error email to {user_email} for task {task.id}")
except Exception as mail_error:
logger.error(f"Failed to send error email to {user_email} for task {task.id}: {str(mail_error)}")
except Exception as e:
logger.error(f"Error sending error email for task {task.id}: {str(e)}")
@login_required
def download_result(request, task_id):
"""
Download the result file for a completed geocoding task
"""
task = get_object_or_404(GeocodingTask, id=task_id)
# Check permission - user must own the task
if task.user != request.user:
raise PermissionDenied(_("Vous n'avez pas accès à ce fichier."))
# Check if task is completed and has output file
if task.status != 'completed' or not task.output_file:
raise Http404(_("Le fichier résultat n'est pas disponible."))
# Serve the file
if not os.path.exists(task.output_file.path):
raise Http404(_("Le fichier résultat n'a pas été trouvé."))
# FileResponse will handle closing the file when done
response = FileResponse(
open(task.output_file.path, 'rb'),
content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
filename=os.path.basename(task.output_file.name)
)
return response
@login_required
def geolocation_tasks_status(request):
"""
API endpoint to get the status of user's geocoding tasks.
Used for auto-refresh functionality.
Only returns tasks from the last 24 hours to limit payload size.
"""
# Only get tasks from the last 24 hours
last_24h = timezone.now() - timezone.timedelta(hours=24)
user_tasks = GeocodingTask.objects.filter(
user=request.user,
created_at__gte=last_24h
).order_by('-created_at')[:10]
tasks_data = []
for task in user_tasks:
# Get filename from uploaded_file
filename = os.path.basename(task.uploaded_file.name) if task.uploaded_file else "-"
# Build download URL if task is completed
download_url = None
if task.status == 'completed' and task.output_file:
download_url = reverse('schools:download_result', args=[task.id])
tasks_data.append({
'id': task.id,
'filename': filename,
'created_at': task.created_at.strftime('%d/%m/%Y %H:%M') if task.created_at else '-',
'status': task.status,
'total_rows': task.total_rows if task.total_rows else '-',
'successful_geocodes': task.successful_geocodes if task.successful_geocodes else '-',
'error_message': task.error_message if task.error_message else 'Erreur inconnue',
'download_url': download_url,
})
# Check if any task is still pending or processing
has_active_tasks = any(t.status in ['pending', 'processing'] for t in user_tasks)
return JsonResponse({
'tasks': tasks_data,
'has_active_tasks': has_active_tasks,
})

1
streetup/streetup/settings.py
View file

@ -58,7 +58,6 @@ INSTALLED_APPS = [
'mobile.apps.MobileConfig', 'mobile.apps.MobileConfig',
'api.apps.ApiConfig', 'api.apps.ApiConfig',
'sign.apps.SignConfig', 'sign.apps.SignConfig',
'schools.apps.SchoolsConfig',
'maps.apps.MapsConfig', 'maps.apps.MapsConfig',
'analytics.apps.AnalyticsConfig', 'analytics.apps.AnalyticsConfig',
'notifications.apps.NotificationsConfig', 'notifications.apps.NotificationsConfig',

1
streetup/streetup/urls.py
View file

@ -41,7 +41,6 @@ urlpatterns = [
path("mobile/", include("mobile.urls")), path("mobile/", include("mobile.urls")),
path("api/", include("api.urls")), path("api/", include("api.urls")),
path("sign/", include("sign.urls")), path("sign/", include("sign.urls")),
path("schools/", include("schools.urls")),
path("maps/", include("maps.urls")), path("maps/", include("maps.urls")),
path("analytics/", include("analytics.urls")), path("analytics/", include("analytics.urls")),
path("notifications/", include("notifications.urls")), path("notifications/", include("notifications.urls")),