(self, request)
| 2680 | |
| 2681 | @action(detail=False, methods=["post"]) |
| 2682 | def upload(self, request): |
| 2683 | if "file" not in request.FILES: |
| 2684 | return Response( |
| 2685 | {"error": "No file uploaded"}, status=status.HTTP_400_BAD_REQUEST |
| 2686 | ) |
| 2687 | |
| 2688 | file = request.FILES["file"] |
| 2689 | |
| 2690 | # Validate file |
| 2691 | try: |
| 2692 | from dispatcharr.utils import validate_logo_file |
| 2693 | validate_logo_file(file) |
| 2694 | except Exception as e: |
| 2695 | return Response( |
| 2696 | {"error": str(e)}, status=status.HTTP_400_BAD_REQUEST |
| 2697 | ) |
| 2698 | |
| 2699 | # Sanitize filename: strip directory components to prevent path traversal |
| 2700 | try: |
| 2701 | file_path = safe_upload_path(file.name, "/data/logos") |
| 2702 | except ValueError: |
| 2703 | return Response({"error": "Invalid filename."}, status=status.HTTP_400_BAD_REQUEST) |
| 2704 | |
| 2705 | os.makedirs("/data/logos", exist_ok=True) |
| 2706 | with open(file_path, "wb+") as destination: |
| 2707 | for chunk in file.chunks(): |
| 2708 | destination.write(chunk) |
| 2709 | |
| 2710 | # Mark file as processed in Redis to prevent file scanner notifications |
| 2711 | try: |
| 2712 | redis_client = RedisClient.get_client() |
| 2713 | if redis_client: |
| 2714 | # Use the same key format as the file scanner |
| 2715 | redis_key = f"processed_file:{file_path}" |
| 2716 | # Store the actual file modification time to match the file scanner's expectation |
| 2717 | file_mtime = os.path.getmtime(file_path) |
| 2718 | redis_client.setex(redis_key, 60 * 60 * 24 * 3, str(file_mtime)) # 3 day TTL |
| 2719 | logger.debug(f"Marked uploaded logo file as processed in Redis: {file_path} (mtime: {file_mtime})") |
| 2720 | except Exception as e: |
| 2721 | logger.warning(f"Failed to mark logo file as processed in Redis: {e}") |
| 2722 | |
| 2723 | # Get custom name from request data, fallback to filename |
| 2724 | custom_name = request.data.get('name', '').strip() |
| 2725 | logo_name = custom_name if custom_name else os.path.basename(file_path) |
| 2726 | |
| 2727 | logo, _ = Logo.objects.get_or_create( |
| 2728 | url=file_path, |
| 2729 | defaults={ |
| 2730 | "name": logo_name, |
| 2731 | }, |
| 2732 | ) |
| 2733 | |
| 2734 | # Use get_serializer to ensure proper context |
| 2735 | serializer = self.get_serializer(logo) |
| 2736 | return Response( |
| 2737 | serializer.data, |
| 2738 | status=status.HTTP_201_CREATED, |
| 2739 | ) |
nothing calls this directly
no test coverage detected