""" accounts/views/admin_download.py Admin-only views for managing and serving order downloads. Flow: 1. Admin opens dashboard -> sees all orders awaiting download. 2. Admin clicks "Start Download" -> POST to `start_download` -> Celery task queued -> returns JSON with task ID. 3. Admin polls `download_status//` -> progress + status. 4. When status == 'completed', admin clicks "Get Files" -> GET `serve_download//` -> streaming zip response. """ import logging import os from django.http import JsonResponse, StreamingHttpResponse, HttpResponse from django.shortcuts import render, redirect from django.contrib import messages from django.utils import timezone from django.views.decorators.http import require_POST from rest_framework.decorators import action from rest_framework import viewsets from .base import DummySerializer from ..models import LocalUploadedFile, TransferOrder, DownloadTask logger = logging.getLogger(__name__) SERVE_CHUNK_SIZE = 8 * 1024 * 1024 # 8 MB streaming chunks to browser class AdminDownloadViewSet(viewsets.ModelViewSet): serializer_class = DummySerializer # ── Dashboard ───────────────────────────────────────────────────────── @action(detail=False, methods=['get'], url_path=r'') # downloads/ def download_dashboard(self, request): """Render the admin download dashboard.""" if not request.user.is_staff: messages.error(request, "Admin access required") return redirect(f'{request.build_absolute_uri("/")}admin/login/') base_domain = request.build_absolute_uri('/') orders_pending = ( TransferOrder.objects .filter(status__in=['payment_completed', 'storage_selected']) .select_related('user') .prefetch_related('file_selections', 'download_task') .order_by('-created_at') ) orders_queued = ( TransferOrder.objects .filter(status='download_queued') .select_related('user') .prefetch_related('file_selections', 'download_task') .order_by('-updated_at') ) orders_downloading = ( TransferOrder.objects .filter(status='download_in_progress') .select_related('user') .prefetch_related('file_selections', 'download_task') .order_by('-updated_at') ) orders_completed = ( TransferOrder.objects .filter(status__in=['download_completed', 'download_partial', 'completed']) .select_related('user') .prefetch_related('file_selections', 'download_task') .order_by('-updated_at')[:100] ) orders_failed = ( TransferOrder.objects .filter(status='download_failed') .select_related('user') .prefetch_related('file_selections', 'download_task') .order_by('-updated_at') ) context = { 'base_domain': base_domain, 'user': request.user, 'orders_pending': orders_pending, 'orders_queued': orders_queued, 'orders_downloading': orders_downloading, 'orders_completed': orders_completed, 'orders_failed': orders_failed, 'total_pending': orders_pending.count(), 'total_queued': orders_queued.count(), 'total_downloading': orders_downloading.count(), 'total_completed': orders_completed.count(), 'total_failed': orders_failed.count(), } return render(request, 'admin_download_dashboard.html', context) # ── Start download (triggers Celery task) ────────────────────────────── @action(detail=False, methods=['post'], url_path=r'start/(?P[^/.]+)') def start_download(self, request, order_id=None): """ Queue a Celery download task for the given order. Returns JSON so the dashboard can poll for progress. """ if not request.user.is_staff: return JsonResponse({'error': 'Admin access required'}, status=403) try: order = TransferOrder.objects.get(id=order_id) except TransferOrder.DoesNotExist: return JsonResponse({'error': 'Order not found'}, status=404) valid_statuses = [ 'payment_completed', 'storage_selected', 'download_failed', 'download_partial', ] if order.status not in valid_statuses: return JsonResponse({ 'error': f'Order cannot be downloaded in its current state: {order.status}' }, status=400) # Cancel any existing in-progress task for this order try: existing_task = order.download_task if existing_task.celery_task_id and existing_task.status == 'in_progress': from celery.app.control import Control from notcloudstorage.celery import app as celery_app Control(celery_app).revoke(existing_task.celery_task_id, terminate=False) except DownloadTask.DoesNotExist: pass # Queue the task from ..tasks import process_order_download task_result = process_order_download.apply_async( args=[str(order_id)], retry=False, # Task itself handles retries ) # Mark order as queued order.status = 'download_queued' order.download_status = 'Download queued — waiting for worker…' order.retry_count += 1 order.save(update_fields=['status', 'download_status', 'retry_count', 'updated_at']) # Ensure DownloadTask row exists with new task ID DownloadTask.objects.update_or_create( order=order, defaults={ 'celery_task_id': task_result.id, 'status': 'queued', 'progress': 0, 'error_message': '', 'started_at': None, 'completed_at': None, 'temp_zip_path': None, }, ) logger.info("Download task queued | order=%s | celery_id=%s", order_id, task_result.id) return JsonResponse({ 'success': True, 'celery_task_id': task_result.id, 'order_id': str(order.id), 'message': 'Download queued successfully', }) # ── Poll download progress ───────────────────────────────────────────── @action(detail=False, methods=['get'], url_path=r'') # download-status// def download_status(self, request, order_id=None): """Return current download progress as JSON. Poll every ~2 seconds from frontend.""" if not request.user.is_staff: return JsonResponse({'error': 'Admin access required'}, status=403) try: order = TransferOrder.objects.prefetch_related('file_selections').get(id=order_id) except TransferOrder.DoesNotExist: return JsonResponse({'error': 'Order not found'}, status=404) # Per-file progress file_list = [] for fs in order.file_selections.all(): file_list.append({ 'id': str(fs.id), 'name': fs.file_name, 'final_name': fs.downloaded_filename or fs.file_name, 'status': fs.download_status, 'progress': fs.progress_percentage, 'size_bytes': fs.file_size, 'downloaded_bytes': fs.download_size or 0, 'error': fs.download_error, }) # Task-level progress task_data = { 'status': 'not_started', 'progress': 0, 'error': None, 'zip_ready': False, 'celery_task_id': None, } try: dt = order.download_task task_data = { 'status': dt.status, 'progress': dt.progress, 'error': dt.error_message, 'zip_ready': dt.zip_exists, 'celery_task_id': dt.celery_task_id, 'started_at': dt.started_at.isoformat() if dt.started_at else None, 'completed_at': dt.completed_at.isoformat() if dt.completed_at else None, 'zip_size_bytes': dt.total_size_downloaded, } except DownloadTask.DoesNotExist: pass return JsonResponse({ 'order_id': str(order.id), 'order_number': order.order_number, 'user_email': order.user.email, 'order_status': order.status, 'download_status_msg': order.download_status, 'task': task_data, 'files': file_list, 'total_files': len(file_list), 'downloaded_files': sum(1 for f in file_list if f['status'] == 'downloaded'), 'failed_files': sum(1 for f in file_list if f['status'] == 'failed'), }) # ── Serve completed zip to admin ─────────────────────────────────────── @action(detail=False, methods=['get'], url_path=r'serve/(?P[^/.]+)') #serve// def serve_download(self, request, order_id=None): """ Stream the completed zip file to the admin's browser. Uses a generator to avoid loading the entire archive into memory. """ if not request.user.is_staff: return HttpResponse("Admin access required", status=403) try: order = TransferOrder.objects.select_related('download_task').get(id=order_id) except TransferOrder.DoesNotExist: return HttpResponse("Order not found", status=404) try: dt = order.download_task except DownloadTask.DoesNotExist: return HttpResponse("No download task found for this order", status=404) if dt.status != 'completed': return HttpResponse( f"Download not ready yet. Current status: {dt.status}", status=409 ) if not dt.zip_exists: messages.error(request, "Zip file missing on server — please re-run the download.") return redirect('admin-download-dashboard') zip_path = dt.temp_zip_path zip_size = os.path.getsize(zip_path) zip_filename = os.path.basename(zip_path) def file_iterator(path, chunk_size=SERVE_CHUNK_SIZE): with open(path, 'rb') as f: while True: data = f.read(chunk_size) if not data: break yield data response = StreamingHttpResponse( file_iterator(zip_path), content_type='application/zip', ) response['Content-Disposition'] = f'attachment; filename="{zip_filename}"' response['Content-Length'] = zip_size response['X-Accel-Redirect'] = '' # Prevent nginx buffering the whole file logger.info("Serving zip | order=%s | admin=%s | size=%.1f MB", order_id, request.user.email, zip_size / 1024 / 1024) return response # ── Retry a failed order ─────────────────────────────────────────────── @action(detail=False, methods=['post'], url_path=r'retry/(?P[^/.]+)') def retry_download(self, request, order_id=None): """Reset failed file statuses and re-queue the download task.""" if not request.user.is_staff: return JsonResponse({'error': 'Admin access required'}, status=403) try: order = TransferOrder.objects.get(id=order_id) except TransferOrder.DoesNotExist: return JsonResponse({'error': 'Order not found'}, status=404) if order.status not in ('download_failed', 'download_partial'): return JsonResponse({'error': f'Order status is {order.status} — retry not needed'}, status=400) # Reset only failed files so successfully-downloaded ones are preserved # The task will re-download all files; for large orders, the worker skips # already-downloaded files if the temp dir still exists. order.file_selections.filter(download_status='failed').update( download_status='pending', download_error=None, progress_percentage=0, ) # Re-queue from ..tasks import process_order_download task_result = process_order_download.apply_async(args=[str(order_id)]) order.status = 'download_queued' order.download_status = 'Retry queued…' order.retry_count += 1 order.save(update_fields=['status', 'download_status', 'retry_count', 'updated_at']) DownloadTask.objects.update_or_create( order=order, defaults={ 'celery_task_id': task_result.id, 'status': 'queued', 'progress': 0, 'error_message': '', 'started_at': None, 'completed_at': None, }, ) logger.info("Retry queued | order=%s | celery_id=%s", order_id, task_result.id) return JsonResponse({'success': True, 'celery_task_id': task_result.id}) # ── Cancel an in-progress task ───────────────────────────────────────── @action(detail=False, methods=['post'], url_path=r'cancel/(?P[^/.]+)') # cancel-download// def cancel_download(self, request, order_id=None): """Revoke the Celery task and reset the order.""" if not request.user.is_staff: return JsonResponse({'error': 'Admin access required'}, status=403) try: order = TransferOrder.objects.select_related('download_task').get(id=order_id) dt = order.download_task except (TransferOrder.DoesNotExist, DownloadTask.DoesNotExist): return JsonResponse({'error': 'Order or task not found'}, status=404) if dt.celery_task_id: try: from notcloudstorage.celery import app as celery_app # adjust import celery_app.control.revoke(dt.celery_task_id, terminate=True, signal='SIGTERM') except Exception as exc: logger.warning("Failed to revoke task %s: %s", dt.celery_task_id, exc) dt.status = 'failed' dt.error_message = 'Cancelled by admin' dt.save(update_fields=['status', 'error_message', 'updated_at']) order.status = 'download_failed' order.download_status = 'Cancelled by admin' order.save(update_fields=['status', 'download_status', 'updated_at']) return JsonResponse({'success': True, 'message': 'Task cancelled'}) # --- Delete the zip files _______________________________________- @action(detail=False, methods=['post'], url_path=r'') #delete-zip/(?P[^/.]+) def delete_zip(self, request, order_id=None): """Delete the zip file from server for a completed order.""" if not request.user.is_staff: return JsonResponse({'error': 'Admin access required'}, status=403) try: order = TransferOrder.objects.select_related('download_task').get(id=order_id) dt = order.download_task except (TransferOrder.DoesNotExist, DownloadTask.DoesNotExist): return JsonResponse({'error': 'Order or download task not found'}, status=404) if dt.status != 'completed': return JsonResponse({'error': 'Download not completed yet'}, status=400) if not dt.temp_zip_path or not os.path.exists(dt.temp_zip_path): return JsonResponse({'error': 'Zip file not found on server'}, status=404) try: os.remove(dt.temp_zip_path) dt.temp_zip_path = None dt.save(update_fields=['temp_zip_path']) logger.info("Deleted zip for order %s", order_id) return JsonResponse({'success': True, 'message': 'Zip deleted'}) except Exception as e: return JsonResponse({'error': str(e)}, status=500) # ── Single local file download (utility) ────────────────────────────── @action(detail=False, methods=['get'], url_path=r'local-file/(?P[^/.]+)') def download_local_file(self, request, file_id=None): """Stream a single locally-uploaded file to admin.""" if not request.user.is_staff: return JsonResponse({'error': 'Admin access required'}, status=403) try: uf = LocalUploadedFile.objects.get(id=file_id) except LocalUploadedFile.DoesNotExist: return HttpResponse("File not found", status=404) if not uf.file or not os.path.exists(uf.file.path): return HttpResponse("File not found on server", status=404) def _iter(path): with open(path, 'rb') as f: while chunk := f.read(SERVE_CHUNK_SIZE): yield chunk response = StreamingHttpResponse(_iter(uf.file.path), content_type='application/octet-stream') response['Content-Disposition'] = f'attachment; filename="{uf.original_name}"' response['Content-Length'] = uf.file_size return response # ---- VIEW FILES ---------- @action(detail=False, methods=['get'], url_path=r'') # order-files// def get_order_files(self, request, order_id=None): """Get all files for a specific order""" print(f"=== GETTING ORDER FILES: {order_id} ===") # if not request.user.is_staff: # print(" UNAUTHORIZED: Non-staff user") # return JsonResponse({'error': 'Admin access required'}, status=403) try: from ..models import TransferOrder order = TransferOrder.objects.get(id=order_id) file_selections = order.file_selections.all().order_by('file_name') files_data = [] for file_selection in file_selections: files_data.append({ 'id': str(file_selection.id), 'name': file_selection.file_name, 'size': file_selection.file_size, 'download_status': file_selection.download_status, 'download_error': file_selection.download_error, 'downloaded_at': file_selection.downloaded_at.isoformat() if file_selection.downloaded_at else None, 'download_size': file_selection.download_size, 'selected': False }) print(f" ORDER FILES RETRIEVED: {len(files_data)} files") return JsonResponse({ 'order_id': str(order.id), 'order_number': order.order_number, 'user_email': order.user.email, 'cloud_source': order.cloud_source,'estimated_completion': order.estimated_completion.isoformat() if order.estimated_completion else None, 'total_files': len(files_data), 'files': files_data }) except TransferOrder.DoesNotExist: print(f" ORDER NOT FOUND: {order_id}") return JsonResponse({'error': 'Order not found'}, status=404) except Exception as e: print(f" GET ORDER FILES FAILED: {str(e)}") return JsonResponse({'error': f'Failed to get order files: {str(e)}'}, status=500) @action(detail=False, methods=['post'], url_path='') # download-files/ def download_selected_files(self, request): """ Queue a Celery download task for the given order. Accepts: order_id (required) file_ids[] (optional) – list of FileSelection IDs to download Returns JSON for AJAX (used by the dashboard). """ if not request.user.is_staff: return JsonResponse({'error': 'Admin access required'}, status=403) order_id = request.POST.get('order_id') file_ids = request.POST.getlist('file_ids[]') if not order_id: return JsonResponse({'error': 'Missing order_id'}, status=400) try: order = TransferOrder.objects.get(id=order_id) except TransferOrder.DoesNotExist: return JsonResponse({'error': 'Order not found'}, status=404) # Validate order status if order.status not in ['payment_completed', 'storage_selected', 'download_failed', 'download_partial']: return JsonResponse({'error': f'Order cannot be downloaded (status={order.status})'}, status=400) # Optionally filter files (if file_ids provided) if file_ids: files_to_download = order.file_selections.filter(id__in=file_ids) if not files_to_download.exists(): return JsonResponse({'error': 'No valid files selected'}, status=400) else: files_to_download = order.file_selections.all() file_ids = [str(fs.id) for fs in files_to_download] # all files # Queue Celery task (pass file_ids as argument) from ..tasks import process_order_download task_result = process_order_download.apply_async( args=[str(order.id)], kwargs={'file_ids': file_ids}, retry=False ) # Update order status order.status = 'download_queued' order.download_status = f'Queued – {len(file_ids)} file(s)' order.save(update_fields=['status', 'download_status', 'updated_at']) # Create/update DownloadTask record DownloadTask.objects.update_or_create( order=order, defaults={ 'celery_task_id': task_result.id, 'status': 'queued', 'progress': 0, 'error_message': '', 'started_at': None, 'completed_at': None, 'temp_zip_path': None, }, ) return JsonResponse({ 'success': True, 'celery_task_id': task_result.id, 'order_id': str(order.id), 'message': f'Download queued for {len(file_ids)} file(s)' })