427 lines
16 KiB
Python
427 lines
16 KiB
Python
"""
|
|
API views for asset history.
|
|
"""
|
|
from django.http import JsonResponse
|
|
from django.views.decorators.http import require_GET
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.utils.translation import gettext as _
|
|
from django.utils.dateparse import parse_datetime, parse_date
|
|
from django.utils import timezone
|
|
from django.contrib.auth.decorators import login_required
|
|
from functools import wraps
|
|
from assets.models import AssetHistory
|
|
from assets.views.asset_editing import _get_asset_by_ct
|
|
from assets.permissions import get_user_asset_thematic_access, _get_asset_thematic
|
|
from common.models import UserConfig
|
|
from datetime import datetime
|
|
|
|
|
|
def _check_user_thematic_access(user, asset, require_edit=False):
|
|
"""
|
|
Check if user has access to view an asset's history.
|
|
Returns (has_access, error_message) tuple.
|
|
"""
|
|
if not user.is_authenticated:
|
|
return False, JsonResponse({
|
|
'success': False,
|
|
'error': _("Authentication required")
|
|
}, status=401)
|
|
|
|
# Get asset's thematic
|
|
thematic = _get_asset_thematic(asset)
|
|
if not thematic:
|
|
return False, JsonResponse({
|
|
'success': False,
|
|
'error': _("Cannot determine asset's thematic")
|
|
}, status=400)
|
|
|
|
# Check user's permissions for this thematic
|
|
can_view, can_edit = get_user_asset_thematic_access(user, thematic)
|
|
|
|
if require_edit and not can_edit:
|
|
return False, JsonResponse({
|
|
'success': False,
|
|
'error': _("You do not have permission to edit assets for this thematic")
|
|
}, status=403)
|
|
|
|
if not can_view:
|
|
return False, JsonResponse({
|
|
'success': False,
|
|
'error': _("You do not have permission to view assets for this thematic")
|
|
}, status=403)
|
|
|
|
return True, None
|
|
|
|
|
|
def _get_user_accessible_thematics(user):
|
|
"""
|
|
Get all thematics accessible to a user.
|
|
Returns list of Thematic objects.
|
|
"""
|
|
try:
|
|
user_config = UserConfig.objects.get(user=user)
|
|
user_thematics = user_config.userthematics.all()
|
|
return [ut.thematic for ut in user_thematics if ut.can_view_assets]
|
|
except UserConfig.DoesNotExist:
|
|
return []
|
|
|
|
|
|
def _short_name(user):
|
|
"""Return 'First L.' style short name for user."""
|
|
if not user:
|
|
return ""
|
|
first = getattr(user, "first_name", "") or ""
|
|
last = getattr(user, "last_name", "") or ""
|
|
initial = (last[0].upper() + ".") if last else ""
|
|
return f"{first} {initial}".strip()
|
|
|
|
|
|
def history_view_required(require_edit=False):
|
|
"""
|
|
Decorator to check if user has permission to access history for a specific asset.
|
|
"""
|
|
def decorator(view_func):
|
|
@wraps(view_func)
|
|
def wrapper(request, *args, **kwargs):
|
|
if not request.user.is_authenticated:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': _("Authentication required")
|
|
}, status=401)
|
|
|
|
# The view will handle specific asset access checks
|
|
return view_func(request, *args, **kwargs)
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
@require_GET
|
|
@history_view_required()
|
|
def get_asset_history(request, asset_model, asset_id, thematic_code=None):
|
|
"""
|
|
Get the history of changes for a specific asset.
|
|
Returns JSON with list of changes ordered by date (newest first).
|
|
Optional thematic_code parameter for URL structure compatibility.
|
|
|
|
Permissions: User must have view_assets permission for the asset's thematic.
|
|
"""
|
|
# Check authentication first
|
|
if not request.user.is_authenticated:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': _("Authentication required")
|
|
}, status=401)
|
|
|
|
try:
|
|
asset, model_class, content_type = _get_asset_by_ct(asset_model, asset_id)
|
|
|
|
# Verify asset belongs to specified thematic if provided
|
|
if thematic_code and hasattr(asset, 'thematic'):
|
|
if not asset.thematic or asset.thematic.code != thematic_code:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': f"Asset does not belong to thematic '{thematic_code}'"
|
|
}, status=404)
|
|
except Exception as e:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=404)
|
|
|
|
# Check user permissions for this asset
|
|
has_access, error_response = _check_user_thematic_access(request.user, asset, require_edit=False)
|
|
if not has_access:
|
|
return error_response
|
|
|
|
# Get history for this asset
|
|
history = AssetHistory.objects.filter(
|
|
content_type=content_type,
|
|
object_id=asset_id
|
|
).select_related('user').order_by('-timestamp')
|
|
|
|
# Format data for response
|
|
history_data = []
|
|
for entry in history:
|
|
history_data.append({
|
|
'id': entry.id,
|
|
'timestamp': entry.timestamp.isoformat(),
|
|
'user': entry.user.get_full_name() if entry.user else _("Système"),
|
|
'user_short_name': _short_name(entry.user) if entry.user else _("Système"),
|
|
'user_username': entry.user.username if entry.user else None,
|
|
'field_name': entry.field_name,
|
|
'field_display_name': entry.get_field_display_name(),
|
|
'old_value': entry.old_value,
|
|
'new_value': entry.new_value,
|
|
'old_value_display': entry.get_display_old_value(),
|
|
'new_value_display': entry.get_display_new_value(),
|
|
'action_type': entry.action_type,
|
|
'action_type_display': entry.get_action_type_display(),
|
|
})
|
|
|
|
return JsonResponse({
|
|
'success': True,
|
|
'history': history_data,
|
|
'count': len(history_data)
|
|
})
|
|
|
|
|
|
@require_GET
|
|
@history_view_required()
|
|
def get_global_history(request, thematic_code=None):
|
|
"""
|
|
Get recent history across all assets (for timeline views).
|
|
Supports pagination and filtering.
|
|
Optional thematic_code parameter to filter by thematic.
|
|
|
|
Permissions:
|
|
- If thematic_code is specified, user must have view_assets permission for that thematic.
|
|
- If no thematic_code, user can only see history for thematics they have access to.
|
|
"""
|
|
# Get query parameters
|
|
limit = int(request.GET.get('limit', 50))
|
|
offset = int(request.GET.get('offset', 0))
|
|
thematic = thematic_code or request.GET.get('thematic', None)
|
|
action_type = request.GET.get('action_type', None)
|
|
user_id = request.GET.get('user_id', None)
|
|
|
|
# Ensure limits are reasonable
|
|
limit = min(limit, 5000) # Cap at 5000 to prevent huge queries
|
|
offset = max(offset, 0)
|
|
|
|
# Get user's accessible thematics
|
|
accessible_thematics = _get_user_accessible_thematics(request.user)
|
|
if not accessible_thematics:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': _("You do not have access to any thematics")
|
|
}, status=403)
|
|
|
|
# Base query - only include content types from accessible thematics
|
|
from common.models import Thematic
|
|
accessible_thematic_codes = [t.code for t in accessible_thematics]
|
|
|
|
# Get all asset categories for accessible thematics
|
|
asset_content_type_ids = []
|
|
for thematic_obj in accessible_thematics:
|
|
categories = thematic_obj.asset_categories.all()
|
|
for category in categories:
|
|
if hasattr(category, 'asset_models'):
|
|
content_types = ContentType.objects.filter(
|
|
model__in=[m.lower() for m in category.asset_models]
|
|
)
|
|
asset_content_type_ids.extend([ct.id for ct in content_types])
|
|
|
|
query = AssetHistory.objects.filter(content_type_id__in=asset_content_type_ids)
|
|
|
|
# Apply additional filters
|
|
if thematic:
|
|
# User is requesting a specific thematic - verify they have access
|
|
try:
|
|
thematic_obj = Thematic.objects.get(code=thematic)
|
|
if thematic_obj not in accessible_thematics:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': _("You do not have permission to view this thematic")
|
|
}, status=403)
|
|
|
|
categories = thematic_obj.asset_categories.all()
|
|
thematic_content_type_ids = []
|
|
for category in categories:
|
|
if hasattr(category, 'asset_models'):
|
|
content_types = ContentType.objects.filter(
|
|
model__in=[m.lower() for m in category.asset_models]
|
|
)
|
|
thematic_content_type_ids.extend([ct.id for ct in content_types])
|
|
|
|
if thematic_content_type_ids:
|
|
query = query.filter(content_type_id__in=thematic_content_type_ids)
|
|
except Thematic.DoesNotExist:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': _("Thematic not found")
|
|
}, status=404)
|
|
|
|
if action_type:
|
|
query = query.filter(action_type=action_type)
|
|
|
|
if user_id:
|
|
query = query.filter(user_id=user_id)
|
|
|
|
# Get total count
|
|
total_count = query.count()
|
|
|
|
# Apply pagination
|
|
history = query.select_related('user', 'content_type').order_by('-timestamp')[offset:offset + limit]
|
|
|
|
# Format data
|
|
history_data = []
|
|
for entry in history:
|
|
# Get asset info
|
|
asset_info = None
|
|
try:
|
|
asset = entry.asset
|
|
if asset:
|
|
asset_info = {
|
|
'code': asset.code,
|
|
'name': asset.get_name() if hasattr(asset, 'get_name') else str(asset),
|
|
'model': entry.content_type.model,
|
|
'id': asset.id,
|
|
}
|
|
except Exception:
|
|
# Asset may have been deleted; fallback to identifiers only
|
|
asset_info = {
|
|
'model': entry.content_type.model,
|
|
'id': entry.object_id,
|
|
}
|
|
|
|
history_data.append({
|
|
'id': entry.id,
|
|
'timestamp': entry.timestamp.isoformat(),
|
|
'user': entry.user.get_full_name() if entry.user else _("Système"),
|
|
'user_short_name': _short_name(entry.user) if entry.user else _("Système"),
|
|
'user_username': entry.user.username if entry.user else None,
|
|
'field_name': entry.field_name,
|
|
'field_display_name': entry.get_field_display_name(),
|
|
'old_value_display': entry.get_display_old_value(),
|
|
'new_value_display': entry.get_display_new_value(),
|
|
'action_type': entry.action_type,
|
|
'action_type_display': entry.get_action_type_display(),
|
|
'asset': asset_info,
|
|
})
|
|
|
|
return JsonResponse({
|
|
'success': True,
|
|
'history': history_data,
|
|
'count': len(history_data),
|
|
'total': total_count,
|
|
'has_more': (offset + limit) < total_count
|
|
})
|
|
|
|
|
|
@require_GET
|
|
@history_view_required()
|
|
def get_asset_status_at_date(request, asset_model, asset_id, thematic_code=None):
|
|
"""
|
|
Get the status of an asset at a specific date.
|
|
Query params:
|
|
- date (ISO format, e.g., 2025-01-23 or 2025-01-23T15:30:00)
|
|
- app_label (optional, to disambiguate model names, e.g., 'assets' or 'sign')
|
|
URL params:
|
|
- thematic_code (optional, for URL structure, validates asset belongs to thematic)
|
|
|
|
Permissions: User must have view_assets permission for the asset's thematic.
|
|
"""
|
|
# Check authentication first
|
|
if not request.user.is_authenticated:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': _("Authentication required")
|
|
}, status=401)
|
|
|
|
# Parse date parameter
|
|
date_str = request.GET.get('date')
|
|
if not date_str:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': _("Paramètre 'date' manquant")
|
|
}, status=400)
|
|
|
|
try:
|
|
# Try to parse as datetime first, then as date
|
|
try:
|
|
target_date = parse_datetime(date_str)
|
|
if not target_date:
|
|
parsed_date = parse_date(date_str)
|
|
if not parsed_date:
|
|
raise ValueError("Invalid date format")
|
|
target_date = datetime.combine(parsed_date, datetime.min.time())
|
|
target_date = timezone.make_aware(target_date)
|
|
except:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': _("Format de date invalide. Utilisez ISO 8601.")
|
|
}, status=400)
|
|
except Exception as e:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=400)
|
|
|
|
# Get optional app_label to disambiguate if multiple models with same name exist
|
|
app_label = request.GET.get('app_label')
|
|
|
|
try:
|
|
# Try to resolve the asset model (handle multiple ContentType matches)
|
|
if app_label:
|
|
content_type = ContentType.objects.get(app_label=app_label, model=asset_model.lower())
|
|
else:
|
|
# First try to get via _get_asset_by_ct (backward compatible)
|
|
try:
|
|
asset, model_class, content_type = _get_asset_by_ct(asset_model, asset_id)
|
|
except Exception:
|
|
# If fails, try to find the model
|
|
content_types = list(ContentType.objects.filter(model=asset_model.lower()))
|
|
if not content_types:
|
|
raise ContentType.DoesNotExist(f"Model {asset_model} not found")
|
|
if len(content_types) > 1:
|
|
# Multiple matches - return error with options
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': f"Multiple models found: {', '.join([f'{ct.app_label}.{ct.model}' for ct in content_types])}. Specify app_label parameter.",
|
|
'options': [{'app_label': ct.app_label, 'model': ct.model} for ct in content_types]
|
|
}, status=400)
|
|
content_type = content_types[0]
|
|
except ContentType.DoesNotExist as e:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=404)
|
|
except Exception as e:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=404)
|
|
|
|
# Get the asset to verify it exists
|
|
try:
|
|
asset_model_class = content_type.model_class()
|
|
asset = asset_model_class.objects.get(pk=asset_id)
|
|
|
|
# Verify asset belongs to specified thematic if provided
|
|
if thematic_code and hasattr(asset, 'thematic'):
|
|
if not asset.thematic or asset.thematic.code != thematic_code:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': f"Asset does not belong to thematic '{thematic_code}'"
|
|
}, status=404)
|
|
except asset_model_class.DoesNotExist:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': f"{asset_model} with id {asset_id} not found"
|
|
}, status=404)
|
|
except Exception as e:
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': str(e)
|
|
}, status=404)
|
|
|
|
# Check user permissions for this asset
|
|
has_access, error_response = _check_user_thematic_access(request.user, asset, require_edit=False)
|
|
if not has_access:
|
|
return error_response
|
|
|
|
# Get status at date
|
|
status = AssetHistory.objects.get_asset_status_at_date(asset, target_date)
|
|
|
|
from assets.models.core import ASSET_STATUS_CHOICES
|
|
status_display_dict = dict(ASSET_STATUS_CHOICES)
|
|
|
|
return JsonResponse({
|
|
'success': True,
|
|
'asset_id': asset.id,
|
|
'asset_code': asset.code,
|
|
'asset_model': f"{content_type.app_label}.{content_type.model}",
|
|
'date': target_date.isoformat(),
|
|
'status': status,
|
|
'status_display': status_display_dict.get(status, _('Inexistant')) if status else _('Inexistant')
|
|
})
|