Compare commits

..

1 Commits

Author SHA1 Message Date
Will Miao
ae7bfdb517 fix(download): normalize civitai.red download URLs (#898) 2026-04-16 18:25:16 +08:00
4 changed files with 50 additions and 5 deletions

View File

@@ -16,7 +16,7 @@ from ..utils.constants import (
SUPPORTED_DOWNLOAD_SKIP_BASE_MODELS, SUPPORTED_DOWNLOAD_SKIP_BASE_MODELS,
VALID_LORA_TYPES, VALID_LORA_TYPES,
) )
from ..utils.civitai_utils import rewrite_preview_url from ..utils.civitai_utils import normalize_civitai_download_url, rewrite_preview_url
from ..utils.preview_selection import resolve_mature_threshold, select_preview_media from ..utils.preview_selection import resolve_mature_threshold, select_preview_media
from ..utils.utils import sanitize_folder_name from ..utils.utils import sanitize_folder_name
from ..utils.exif_utils import ExifUtils from ..utils.exif_utils import ExifUtils
@@ -644,7 +644,9 @@ class DownloadManager:
if mirrors: if mirrors:
for mirror in mirrors: for mirror in mirrors:
if mirror.get("deletedAt") is None and mirror.get("url"): if mirror.get("deletedAt") is None and mirror.get("url"):
download_urls.append(mirror["url"]) download_urls.append(
normalize_civitai_download_url(mirror["url"])
)
# When source is 'civarchive', prioritize non-Civitai URLs # When source is 'civarchive', prioritize non-Civitai URLs
# This avoids failed downloads from deleted Civitai models # This avoids failed downloads from deleted Civitai models
@@ -663,7 +665,9 @@ class DownloadManager:
else: else:
download_url = file_info.get("downloadUrl") download_url = file_info.get("downloadUrl")
if download_url: if download_url:
download_urls.append(download_url) download_urls.append(
normalize_civitai_download_url(download_url)
)
if not download_urls: if not download_urls:
return {"success": False, "error": "No mirror URL found"} return {"success": False, "error": "No mirror URL found"}
@@ -1138,6 +1142,7 @@ class DownloadManager:
pause_control.update_stall_timeout(downloader.stall_timeout) pause_control.update_stall_timeout(downloader.stall_timeout)
last_error = None last_error = None
for download_url in download_urls: for download_url in download_urls:
download_url = normalize_civitai_download_url(download_url)
use_auth = download_url.startswith(CIVITAI_DOWNLOAD_URL_PREFIXES) use_auth = download_url.startswith(CIVITAI_DOWNLOAD_URL_PREFIXES)
download_kwargs = { download_kwargs = {
"progress_callback": lambda progress, snapshot=None: ( "progress_callback": lambda progress, snapshot=None: (

View File

@@ -119,6 +119,24 @@ def extract_civitai_image_id(url: str | None) -> str | None:
return path_match.group(1) return path_match.group(1)
def normalize_civitai_download_url(url: str | None) -> str | None:
"""Rewrite Civitai download URLs to the canonical authenticated host."""
if not url:
return url
try:
parsed = urlparse(url)
except ValueError:
return url
hostname = parsed.hostname.lower() if parsed.hostname else None
if hostname != "civitai.red" or not parsed.path.startswith("/api/download/"):
return url
return urlunparse(parsed._replace(netloc="civitai.com"))
def extract_civitai_page_host(url: str | None) -> str | None: def extract_civitai_page_host(url: str | None) -> str | None:
"""Extract the supported Civitai page host from a URL.""" """Extract the supported Civitai page host from a URL."""

View File

@@ -369,8 +369,8 @@ async def test_execute_download_uses_auth_for_red_civitai_downloads(monkeypatch,
) )
assert result == {"success": True} assert result == {"success": True}
assert recorded_use_auth == [("https://civitai.red/api/download/models/119514", True)] assert recorded_use_auth == [("https://civitai.com/api/download/models/119514", True)]
assert "https://civitai.red/api/download/".startswith(CIVITAI_DOWNLOAD_URL_PREFIXES) assert "https://civitai.com/api/download/".startswith(CIVITAI_DOWNLOAD_URL_PREFIXES)
@pytest.mark.asyncio @pytest.mark.asyncio

View File

@@ -3,6 +3,7 @@ from py.utils.civitai_utils import (
extract_civitai_image_id, extract_civitai_image_id,
extract_civitai_model_url_parts, extract_civitai_model_url_parts,
is_supported_civitai_page_host, is_supported_civitai_page_host,
normalize_civitai_download_url,
resolve_license_info, resolve_license_info,
resolve_license_payload, resolve_license_payload,
) )
@@ -122,3 +123,24 @@ def test_extract_civitai_image_id_supports_red():
def test_extract_civitai_image_id_rejects_non_civitai_host(): def test_extract_civitai_image_id_rejects_non_civitai_host():
assert extract_civitai_image_id("https://example.com/images/126920345") is None assert extract_civitai_image_id("https://example.com/images/126920345") is None
def test_normalize_civitai_download_url_rewrites_red_to_com():
url = "https://civitai.red/api/download/models/2786889?type=Model&format=SafeTensor"
assert (
normalize_civitai_download_url(url)
== "https://civitai.com/api/download/models/2786889?type=Model&format=SafeTensor"
)
def test_normalize_civitai_download_url_keeps_non_download_red_urls():
url = "https://civitai.red/models/65423/nijimecha-artstyle?modelVersionId=777"
assert normalize_civitai_download_url(url) == url
def test_normalize_civitai_download_url_keeps_existing_com_urls():
url = "https://civitai.com/api/download/models/2786889?type=Model&format=SafeTensor"
assert normalize_civitai_download_url(url) == url