MCPcopy Create free account
hub / github.com/Dispatcharr/Dispatcharr / upload

Method upload

apps/channels/api_views.py:2682–2739  ·  view source on GitHub ↗
(self, request)

Source from the content-addressed store, hash-verified

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 )

Callers

nothing calls this directly

Calls 4

validate_logo_fileFunction · 0.90
safe_upload_pathFunction · 0.90
get_clientMethod · 0.80
getMethod · 0.45

Tested by

no test coverage detected