Streams the logo file, whether it's local or remote.
(self, request, pk=None)
| 2740 | |
| 2741 | @action(detail=True, methods=["get"], permission_classes=[AllowAny]) |
| 2742 | def cache(self, request, pk=None): |
| 2743 | """Streams the logo file, whether it's local or remote.""" |
| 2744 | logo = self.get_object() |
| 2745 | logo_url = logo.url |
| 2746 | if logo_url.startswith("/data"): # Local file |
| 2747 | if not os.path.exists(logo_url): |
| 2748 | raise Http404("Image not found") |
| 2749 | stat = os.stat(logo_url) |
| 2750 | # Get proper mime type (first item of the tuple) |
| 2751 | content_type, _ = mimetypes.guess_type(logo_url) |
| 2752 | if not content_type: |
| 2753 | content_type = "image/jpeg" # Default to a common image type |
| 2754 | |
| 2755 | # Use context manager and set Content-Disposition to inline |
| 2756 | response = StreamingHttpResponse( |
| 2757 | open(logo_url, "rb"), content_type=content_type |
| 2758 | ) |
| 2759 | response["Cache-Control"] = "public, max-age=14400" # Cache in browser for 4 hours |
| 2760 | response["Last-Modified"] = http_date(stat.st_mtime) |
| 2761 | response["Content-Disposition"] = 'inline; filename="{}"'.format( |
| 2762 | os.path.basename(logo_url) |
| 2763 | ) |
| 2764 | return response |
| 2765 | |
| 2766 | else: # Remote image |
| 2767 | # Skip URLs that recently failed to avoid blocking workers |
| 2768 | # on unreachable hosts (e.g., dead CDNs referenced by old recordings). |
| 2769 | fail_expiry = _logo_fetch_failures.get(logo_url) |
| 2770 | if fail_expiry and time.monotonic() < fail_expiry: |
| 2771 | raise Http404("Remote image temporarily unavailable") |
| 2772 | |
| 2773 | try: |
| 2774 | # Get the default user agent |
| 2775 | try: |
| 2776 | default_user_agent_id = CoreSettings.get_default_user_agent_id() |
| 2777 | user_agent_obj = UserAgent.objects.get(id=int(default_user_agent_id)) |
| 2778 | user_agent = user_agent_obj.user_agent |
| 2779 | except (CoreSettings.DoesNotExist, UserAgent.DoesNotExist, ValueError): |
| 2780 | # Fallback if default not found |
| 2781 | from core.utils import dispatcharr_user_agent |
| 2782 | user_agent = dispatcharr_user_agent() |
| 2783 | |
| 2784 | # Hard total timeout (connect + full download) prevents a slow |
| 2785 | # server dripping bytes from holding a greenlet indefinitely. |
| 2786 | _LOGO_TOTAL_TIMEOUT = 10 # seconds |
| 2787 | _LOGO_MAX_BYTES = 5 * 1024 * 1024 # 5 MB |
| 2788 | |
| 2789 | remote_response = requests.get( |
| 2790 | logo_url, |
| 2791 | stream=True, |
| 2792 | timeout=(3, 5), # (connect_timeout, read_timeout per chunk) |
| 2793 | headers={'User-Agent': user_agent} |
| 2794 | ) |
| 2795 | if remote_response.status_code == 200: |
| 2796 | # Eagerly read the full image with a total time + size cap |
| 2797 | # so the greenlet is released quickly. |
| 2798 | chunks = [] |
| 2799 | total = 0 |
nothing calls this directly
no test coverage detected