199 lines
7.2 KiB
Python
199 lines
7.2 KiB
Python
"""
|
|
Management command to query assets by status at a specific date.
|
|
Usage: python manage.py assets_at_date <asset_model> <status> <date>
|
|
Examples:
|
|
python manage.py assets_at_date signpole active 2025-01-15
|
|
python manage.py assets_at_date assets.signpole active 2025-01-15 # Specify app
|
|
python manage.py assets_at_date sign.signpole active 2025-01-15 # Specify app
|
|
"""
|
|
|
|
from django.core.management.base import BaseCommand
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.utils.dateparse import parse_datetime, parse_date
|
|
from django.utils import timezone
|
|
from datetime import datetime
|
|
from assets.models import AssetHistory
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = 'Query assets by status at a specific date (supports app_label.model_name format)'
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
'asset_model',
|
|
type=str,
|
|
help='The asset model, optionally with app label (e.g., signpole or assets.signpole)'
|
|
)
|
|
parser.add_argument(
|
|
'status',
|
|
type=str,
|
|
nargs='?',
|
|
default=None,
|
|
help='Status to filter by (e.g., active, removed) - optional'
|
|
)
|
|
parser.add_argument(
|
|
'date',
|
|
type=str,
|
|
help='Date in ISO format (e.g., 2025-01-15 or 2025-01-15T12:00:00)'
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
asset_model_input = options['asset_model']
|
|
status_filter = options.get('status')
|
|
date_str = options['date']
|
|
|
|
# Parse 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 Exception as e:
|
|
self.stdout.write(
|
|
self.style.ERROR(f"Invalid date format: {e}")
|
|
)
|
|
return
|
|
|
|
# Parse asset model (may include app_label)
|
|
if '.' in asset_model_input:
|
|
app_label, model_name = asset_model_input.split('.', 1)
|
|
content_types = ContentType.objects.filter(app_label=app_label, model=model_name.lower())
|
|
else:
|
|
model_name = asset_model_input.lower()
|
|
content_types = list(ContentType.objects.filter(model=model_name))
|
|
|
|
# Handle multiple matches
|
|
if not content_types:
|
|
self.stdout.write(
|
|
self.style.ERROR(f"Model '{asset_model_input}' not found")
|
|
)
|
|
return
|
|
|
|
if len(content_types) > 1:
|
|
self.stdout.write(
|
|
self.style.WARNING(f"Multiple models found for '{asset_model_input}':")
|
|
)
|
|
for ct in content_types:
|
|
self.stdout.write(f" - {ct.app_label}.{ct.model}")
|
|
self.stdout.write(
|
|
self.style.WARNING("Using the first one. Specify with 'app_label.model' to choose.\n")
|
|
)
|
|
|
|
content_type = content_types[0]
|
|
asset_model_class = content_type.model_class()
|
|
|
|
self.stdout.write(
|
|
self.style.SUCCESS(
|
|
f"\nAssets of type '{content_type.app_label}.{content_type.model}' on {target_date.date()}"
|
|
)
|
|
)
|
|
if status_filter:
|
|
self.stdout.write(f" Status: {status_filter}\n")
|
|
|
|
# Get all assets created by the date
|
|
created_by_date = list(AssetHistory.objects.filter(
|
|
content_type=content_type,
|
|
action_type='create',
|
|
timestamp__lte=target_date
|
|
).values_list('object_id', flat=True).distinct())
|
|
|
|
if not created_by_date:
|
|
self.stdout.write(self.style.WARNING(" No assets created by this date"))
|
|
return
|
|
|
|
# Get all deletions in one query
|
|
deleted_ids = set(AssetHistory.objects.filter(
|
|
content_type=content_type,
|
|
object_id__in=created_by_date,
|
|
action_type='delete',
|
|
timestamp__lte=target_date
|
|
).values_list('object_id', flat=True))
|
|
|
|
# Get last status for each asset in one query using subquery
|
|
from django.db.models import OuterRef, Subquery
|
|
latest_status = AssetHistory.objects.filter(
|
|
content_type=content_type,
|
|
object_id=OuterRef('object_id'),
|
|
field_name='status',
|
|
timestamp__lte=target_date
|
|
).order_by('-timestamp').values('new_value')[:1]
|
|
|
|
statuses = {}
|
|
status_entries = AssetHistory.objects.filter(
|
|
content_type=content_type,
|
|
object_id__in=created_by_date,
|
|
field_name='status',
|
|
timestamp__lte=target_date
|
|
).values('object_id').annotate(
|
|
last_status=Subquery(latest_status)
|
|
).values('object_id', 'last_status')
|
|
|
|
for entry in status_entries:
|
|
statuses[entry['object_id']] = entry['last_status']
|
|
|
|
# Get all assets in one query for faster access
|
|
assets_dict = {asset.id: asset for asset in asset_model_class.objects.filter(pk__in=created_by_date)}
|
|
|
|
# Build results
|
|
results = []
|
|
for asset_id in created_by_date:
|
|
# Check if deleted
|
|
if asset_id in deleted_ids:
|
|
status = 'DELETED'
|
|
else:
|
|
# Get status from pre-fetched data
|
|
status = statuses.get(asset_id)
|
|
if not status:
|
|
# Fallback to current asset status
|
|
asset_obj = assets_dict.get(asset_id)
|
|
if asset_obj:
|
|
status = getattr(asset_obj, 'status', None)
|
|
|
|
if status_filter and status != status_filter:
|
|
continue
|
|
|
|
# Get asset details from pre-fetched dict
|
|
asset = assets_dict.get(asset_id)
|
|
if asset:
|
|
results.append({
|
|
'id': asset_id,
|
|
'code': asset.code,
|
|
'status': status,
|
|
'name': asset.get_name() if hasattr(asset, 'get_name') else str(asset)
|
|
})
|
|
else:
|
|
results.append({
|
|
'id': asset_id,
|
|
'code': f'ID: {asset_id}',
|
|
'status': status,
|
|
'name': '(Deleted)'
|
|
})
|
|
|
|
# Display results
|
|
if not results:
|
|
self.stdout.write(
|
|
self.style.WARNING(
|
|
f" No assets found with the given criteria"
|
|
)
|
|
)
|
|
return
|
|
|
|
self.stdout.write(f" Found {len(results)} asset(s):\n")
|
|
|
|
for item in results:
|
|
status_color = {
|
|
'active': self.style.SUCCESS,
|
|
'removed': self.style.ERROR,
|
|
'damaged': self.style.WARNING,
|
|
'DELETED': self.style.ERROR,
|
|
}.get(item['status'], self.style.WARNING)
|
|
|
|
status_display = status_color(f"[{item['status']}]")
|
|
self.stdout.write(
|
|
f" • {item['code']:20} {status_display:20} {item['name']}"
|
|
)
|
|
|
|
self.stdout.write(self.style.SUCCESS(f"\n Total: {len(results)} asset(s)\n"))
|