""" 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') })