Upload 7 files

README.md

@@ -4,7 +4,7 @@ emoji: 🤗
 colorFrom: yellow
 colorTo: blue
 sdk: gradio
-sdk_version: 6.11.0
+sdk_version: 6.13.0
 python_version: "3.10.13"
 app_file: app.py
 pinned: false

app.py

@@ -35,10 +35,8 @@ from civitai_to_hf import (
     load_all_civitai,
     refresh_civitai_key_status,
     probe_civitai_api,
-    probe_civitai_green_api,
     update_civitai_gallery_mode,
     begin_probe_civitai,
-    begin_probe_civitai_green,
     clear_retry_state,
     CIVITAI_TYPE,
     CIVITAI_BASEMODEL,
@@ -156,13 +154,12 @@ with gr.Blocks(fill_width=True, delete_cache=(60, 3600)) as demo:
             with gr.Row():
                 use_remaining_button = gr.Button(value="Use Remaining URLs", visible=True, variant="secondary")
                 use_failed_button = gr.Button(value="Retry Failed Only", visible=True, variant="secondary")
-                clear_remaining_button = gr.Button(value="Clear Retry State", visible=True, variant="secondary")
+                clear_remaining_button = gr.Button(value="Clear Remaining", visible=True, variant="secondary")
         with gr.Accordion("Probe / Experimental", open=False, visible=CIVITAI_HF_DEBUG):
             probe_md = gr.Markdown(value="", visible=False)
             with gr.Row():
                 smoke_test_button = gr.Button(value="Smoke Test", variant="secondary")
                 probe_button = gr.Button("Probe Civitai", variant="secondary")
-                probe_green_button = gr.Button("Probe civitai.green", variant="secondary")
         gr.DuplicateButton(value="Duplicate Space")
 
     heavy_outputs = [uploaded_urls, urls_md, urls_remain, urls_failed, civitai_key_status, session_state]
@@ -193,7 +190,7 @@ with gr.Blocks(fill_width=True, delete_cache=(60, 3600)) as demo:
     stop_button.click(fn=None, cancels=[download_evt, smoke_evt], queue=False, api_visibility="undocumented")
     use_remaining_button.click(use_remaining_urls, [urls_remain, dl_url], [dl_url], queue=False, api_visibility="undocumented")
     use_failed_button.click(use_failed_urls, [urls_failed, dl_url], [dl_url], queue=False, api_visibility="undocumented")
-    clear_remaining_button.click(clear_retry_state, [session_state], [urls_md, urls_remain, urls_failed, session_state], queue=False, api_visibility="undocumented")
+    clear_remaining_button.click(clear_retry_state, [session_state], [urls_remain, urls_failed, session_state], queue=False, api_visibility="undocumented")
 
     gr.on(
         triggers=[search_civitai_submit.click, search_civitai_query.submit],
@@ -218,7 +215,6 @@ with gr.Blocks(fill_width=True, delete_cache=(60, 3600)) as demo:
     search_civitai_user.change(refresh_civitai_creators, [search_civitai_user, civitai_key], [search_civitai_user], queue=False, api_visibility="undocumented")
     civitai_key.change(refresh_civitai_key_status, [civitai_key], [civitai_key_status], queue=False, api_visibility="undocumented")
     probe_button.click(begin_probe_civitai, [], [probe_md], queue=False, api_visibility="undocumented").then(probe_civitai_api, [search_civitai_query, civitai_key], [probe_md, civitai_key_status], queue=False, api_visibility="undocumented")
-    probe_green_button.click(begin_probe_civitai_green, [], [probe_md], queue=False, api_visibility="undocumented").then(probe_civitai_green_api, [search_civitai_query, civitai_key], [probe_md, civitai_key_status], queue=False, api_visibility="undocumented")
 
 
 demo.queue()

civitai_to_hf.py

@@ -45,8 +45,13 @@ SEARCH_DETAIL_SIZE = (960, 1280)
 LOAD_ALL_BATCH_SIZE = 8
 NULL_IMAGE_PATH = str(Path(__file__).with_name("null.png"))
 CIVITAI_DEFAULT_ORIGIN = "https://civitai.com"
+CIVITAI_CANONICAL_WEB_ORIGIN = CIVITAI_DEFAULT_ORIGIN
+CIVITAI_RED_ORIGIN = "https://civitai.red"
 CIVITAI_GREEN_ORIGIN = "https://civitai.green"
-CIVITAI_HOST_ALIASES = frozenset({"civitai.com", "www.civitai.com", "civitai.green", "www.civitai.green"})
+CIVITAI_GREEN_HOST_ALIASES = frozenset({"civitai.green", "www.civitai.green"})
+CIVITAI_RED_HOST_ALIASES = frozenset({"civitai.red", "www.civitai.red"})
+CIVITAI_HOST_ALIASES = frozenset({"civitai.com", "www.civitai.com", *CIVITAI_GREEN_HOST_ALIASES, *CIVITAI_RED_HOST_ALIASES})
+CIVITAI_API_ORIGIN_CANDIDATES = (CIVITAI_RED_ORIGIN, CIVITAI_DEFAULT_ORIGIN)
 PREVIEW_VIDEO_EXTS = {".mp4", ".webm", ".mov", ".m4v", ".avi", ".mkv"}
 _FFMPEG_PATH = None
 _FFMPEG_MISSING_LOGGED = False
@@ -59,6 +64,255 @@ RUN_CANCEL_REGISTRY = {}
 RUN_CANCEL_LOCK = threading.Lock()
 CREATOR_SUGGEST_CACHE = {}
 CREATOR_SUGGEST_LOCK = threading.Lock()
+CIVITAI_ACTIVE_API_ORIGIN = ""
+CIVITAI_ACTIVE_API_BASE = ""
+_CIVITAI_API_LOCK = threading.Lock()
+
+
+def canonicalize_civitai_netloc(netloc: str):
+    host = str(netloc or "").strip().lower()
+    if host in CIVITAI_GREEN_HOST_ALIASES or host == "www.civitai.com":
+        return "civitai.com"
+    if host in CIVITAI_RED_HOST_ALIASES:
+        return "civitai.red"
+    return host
+
+
+def canonicalize_civitai_host(netloc: str):
+    return canonicalize_civitai_netloc(netloc)
+
+
+def normalize_civitai_origin(value: str):
+    raw = str(value or "").strip()
+    if not raw:
+        return CIVITAI_DEFAULT_ORIGIN
+    parts = urllib.parse.urlsplit(raw if "://" in raw else f"https://{raw}")
+    host = canonicalize_civitai_netloc(parts.netloc or parts.path)
+    if host in {"civitai.com", "civitai.red"}:
+        return f"https://{host}"
+    return CIVITAI_DEFAULT_ORIGIN
+
+
+def get_civitai_canonical_web_origin():
+    return CIVITAI_CANONICAL_WEB_ORIGIN
+
+
+def get_civitai_display_origin():
+    return get_civitai_canonical_web_origin()
+
+
+def build_civitai_api_base(origin: str):
+    raw = str(origin or "").strip().rstrip("/")
+    return f"{raw}/api/v1" if raw else ""
+
+
+def set_civitai_active_api_origin(origin: str):
+    global CIVITAI_ACTIVE_API_ORIGIN, CIVITAI_ACTIVE_API_BASE
+    normalized = normalize_civitai_origin(origin)
+    if normalized not in CIVITAI_API_ORIGIN_CANDIDATES:
+        normalized = CIVITAI_DEFAULT_ORIGIN
+    base = build_civitai_api_base(normalized)
+    with _CIVITAI_API_LOCK:
+        CIVITAI_ACTIVE_API_ORIGIN = normalized
+        CIVITAI_ACTIVE_API_BASE = base
+    return base
+
+
+def set_civitai_api_origin(origin: str):
+    set_civitai_active_api_origin(origin)
+    return CIVITAI_ACTIVE_API_ORIGIN
+
+
+def get_civitai_api_origin():
+    with _CIVITAI_API_LOCK:
+        return CIVITAI_ACTIVE_API_ORIGIN or ""
+
+
+def probe_civitai_api_origin(session, origin: str, timeout: tuple[float, float] = (3.0, 8.0)):
+    response = None
+    try:
+        base_url = build_civitai_api_base(origin)
+        if not base_url:
+            return False
+        response = session.get(
+            f"{base_url}/tags",
+            params={"limit": 1},
+            headers=get_civitai_headers(""),
+            timeout=timeout,
+        )
+        if not response.ok:
+            return False
+        content_type = str(response.headers.get("content-type") or "").lower()
+        if "json" not in content_type:
+            return False
+        data = response.json()
+        return isinstance(data, dict)
+    except Exception:
+        return False
+    finally:
+        try:
+            if response is not None:
+                response.close()
+        except Exception:
+            pass
+
+
+def get_civitai_active_api_origin(force_refresh: bool = False, session=None):
+    cached_origin = get_civitai_api_origin()
+    if cached_origin and not force_refresh:
+        return cached_origin
+    if session is None:
+        session = create_retry_session(total=4, backoff_factor=0.8)
+    for origin in CIVITAI_API_ORIGIN_CANDIDATES:
+        if probe_civitai_api_origin(session, origin):
+            set_civitai_active_api_origin(origin)
+            return get_civitai_api_origin()
+    set_civitai_active_api_origin(CIVITAI_DEFAULT_ORIGIN)
+    return get_civitai_api_origin()
+
+
+def resolve_civitai_api_origin(session=None):
+    return get_civitai_active_api_origin(session=session)
+
+
+def get_civitai_active_api_base(force_refresh: bool = False, session=None):
+    with _CIVITAI_API_LOCK:
+        cached_base = CIVITAI_ACTIVE_API_BASE or ""
+    if cached_base and not force_refresh:
+        return cached_base
+    get_civitai_active_api_origin(force_refresh=force_refresh, session=session)
+    with _CIVITAI_API_LOCK:
+        return CIVITAI_ACTIVE_API_BASE or build_civitai_api_base(CIVITAI_DEFAULT_ORIGIN)
+
+
+def get_civitai_api_candidate_origins(preferred: str = ""):
+    origins = list(CIVITAI_API_ORIGIN_CANDIDATES)
+    preferred_origin = normalize_civitai_origin(preferred) if preferred else ""
+    if preferred_origin in origins:
+        return [preferred_origin] + [origin for origin in origins if origin != preferred_origin]
+    cached_origin = get_civitai_api_origin()
+    if cached_origin in origins:
+        return [cached_origin] + [origin for origin in origins if origin != cached_origin]
+    return origins
+
+
+def iter_civitai_api_bases(preferred: str = "", session=None):
+    preferred_origin = normalize_civitai_origin(preferred) if preferred else get_civitai_active_api_origin(session=session)
+    return [build_civitai_api_base(origin) for origin in get_civitai_api_candidate_origins(preferred_origin)]
+
+
+def build_civitai_api_url(path: str, origin: str = ""):
+    clean_path = "/" + str(path or "").strip().lstrip("/")
+    if not clean_path.startswith("/api/"):
+        clean_path = f"/api/v1{clean_path}"
+    base_origin = normalize_civitai_origin(origin) if origin else normalize_civitai_origin(get_civitai_api_origin() or CIVITAI_DEFAULT_ORIGIN)
+    return f"{base_origin.rstrip('/')}" + clean_path
+
+
+def build_civitai_model_url(model_id, model_version_id=None):
+    if model_id is None:
+        return ""
+    url = f"{get_civitai_canonical_web_origin().rstrip('/')}/models/{model_id}"
+    if model_version_id is not None:
+        url += f"?modelVersionId={model_version_id}"
+    return url
+
+
+def should_fallback_civitai_api_response(response):
+    if response is None:
+        return True
+    try:
+        status_code = int(getattr(response, "status_code", 0) or 0)
+    except Exception:
+        return True
+    return status_code in {404, 405, 408, 429} or status_code >= 500
+
+
+def request_civitai_api(session, path: str, api_key: str = "", params=None, timeout: tuple[float, float] = (7.0, 30.0),
+                        label: str = "Civitai API", source: str = "civitai-api", preferred_origin: str = "",
+                        non_json_fallback_origin: str = ""):
+    preferred = normalize_civitai_origin(preferred_origin) if preferred_origin else get_civitai_active_api_origin(session=session)
+    non_json_fallback = normalize_civitai_origin(non_json_fallback_origin) if non_json_fallback_origin else ""
+    last_response = None
+    last_exception = None
+    for origin in get_civitai_api_candidate_origins(preferred):
+        url = build_civitai_api_url(path, origin=origin)
+        try:
+            response = civitai_get(session, url, api_key=api_key, params=params, timeout=timeout, label=label, source=source)
+            last_response = response
+            if response is not None and response.ok:
+                try:
+                    get_civitai_response_json(response)
+                except Exception as e:
+                    last_exception = e
+                    log_line("retry", f"{label}: non-json response from {origin}")
+                    try:
+                        response.close()
+                    except Exception:
+                        pass
+                    if non_json_fallback and origin != non_json_fallback:
+                        fallback_url = build_civitai_api_url(path, origin=non_json_fallback)
+                        try:
+                            fallback_response = civitai_get(session, fallback_url, api_key=api_key, params=params, timeout=timeout, label=label, source=source)
+                            last_response = fallback_response
+                            if fallback_response is not None and fallback_response.ok:
+                                get_civitai_response_json(fallback_response)
+                                set_civitai_active_api_origin(non_json_fallback)
+                                return fallback_response
+                            if fallback_response is not None and not should_fallback_civitai_api_response(fallback_response):
+                                return fallback_response
+                            if fallback_response is not None:
+                                fallback_response.close()
+                        except Exception as fallback_error:
+                            last_exception = fallback_error
+                    continue
+                set_civitai_active_api_origin(origin)
+                return response
+            if not should_fallback_civitai_api_response(response):
+                return response
+            if response is not None:
+                response.close()
+        except Exception as e:
+            last_exception = e
+            continue
+    if last_exception is not None:
+        raise last_exception
+    if last_response is not None:
+        return last_response
+    return None
+
+
+_CIVITAI_JSON_MISSING = object()
+
+
+def get_civitai_response_json(response, default=_CIVITAI_JSON_MISSING):
+    if response is None:
+        if default is not _CIVITAI_JSON_MISSING:
+            return default
+        raise ValueError("Civitai response is missing")
+    cached = getattr(response, "_civitai_json", _CIVITAI_JSON_MISSING)
+    if cached is not _CIVITAI_JSON_MISSING:
+        return cached
+    try:
+        payload = response.json()
+    except Exception:
+        if default is not _CIVITAI_JSON_MISSING:
+            return default
+        raise
+    setattr(response, "_civitai_json", payload)
+    return payload
+
+
+def request_civitai_api_url(session, url: str, api_key: str = "", params=None, timeout: tuple[float, float] = (7.0, 30.0),
+                            label: str = "Civitai API", source: str = "civitai-api"):
+    parts = get_civitai_url_parts(url)
+    path = str(parts.path or "")
+    if is_civitai_host(parts.netloc) and path.startswith("/api/"):
+        api_path = path + (f"?{parts.query}" if parts.query else "")
+        return request_civitai_api(session, api_path, api_key=api_key, params=params, timeout=timeout, label=label, source=source)
+    response = civitai_get(session, url, api_key=api_key, params=params, timeout=timeout, label=label, source=source)
+    get_civitai_response_json(response)
+    return response
 
 
 CIVITAI_TYPE = ["Checkpoint", "TextualInversion", "Hypernetwork", "AestheticGradient", "LORA", "LoCon", "DoRA",
@@ -395,7 +649,7 @@ def build_run_outputs(urls, md: str, remain_urls, failed_urls, civitai_key, sess
     final_md = md.strip()
     if final_md:
         final_md += "\n"
-    return gr.update(value=urls, choices=urls), build_run_status_update(session_state), gr.update(value=final_md), remain_update, failed_update, key_status_update, session_state_output(session_state)
+    return gr.update(value=urls, choices=urls), gr.update(value=final_md), remain_update, failed_update, key_status_update, session_state_output(session_state)
 
 
 def set_stage_progress(progress, current: int, total: int, desc: str):
@@ -546,7 +800,7 @@ def get_civitai_url_parts(url: str):
         return urllib.parse.urlsplit("")
 
 def is_civitai_host(netloc: str):
-    return str(netloc or "").strip().lower() in CIVITAI_HOST_ALIASES
+    return canonicalize_civitai_host(netloc) in {"civitai.com", "civitai.red"}
 
 def is_civitai_download_api_path(path: str):
     return re.match(r'^/api/download/models/\d+$', str(path or "").strip()) is not None
@@ -569,7 +823,7 @@ def extract_civitai_model_version_id(url: str):
 def to_civitai_default_download_url(version_id: str, query: str = ""):
     if not str(version_id or "").isdigit():
         return ""
-    base = f"{CIVITAI_DEFAULT_ORIGIN}/api/download/models/{version_id}"
+    base = f"{get_civitai_display_origin()}/api/download/models/{version_id}"
     return f"{base}?{query}" if query else base
 
 def normalize_civitai_download_api_url(url: str):
@@ -583,7 +837,7 @@ def extract_first_civitai_download_url_from_html(html: str):
         return ""
     page = html_lib.unescape(str(html))
     patterns = [
-        r'https?://(?:www\.)?(?:civitai\.com|civitai\.green)/api/download/models/\d+[^\s\'"<>)\]]*',
+        r'https?://(?:www\.)?(?:civitai\.com|civitai\.red|civitai\.green)/api/download/models/\d+[^\s\'"<>)\]]*',
         r"[\"'](/api/download/models/\d+[^\"']*)[\"']",
     ]
     for pattern in patterns:
@@ -612,7 +866,7 @@ def resolve_civitai_model_page_to_download_url(url: str, api_key: str = ""):
     version_id = extract_civitai_model_version_id(raw)
     if version_id:
         return to_civitai_default_download_url(version_id)
-    headers = get_civitai_headers(api_key if parts.netloc.lower().endswith("civitai.com") else "")
+    headers = get_civitai_headers(api_key if canonicalize_civitai_host(parts.netloc) in {"civitai.com", "civitai.red"} else "")
     headers["Referer"] = f"{parts.scheme or 'https'}://{parts.netloc}/"
     session = create_retry_session(total=4, backoff_factor=0.8)
     try:
@@ -643,18 +897,18 @@ def get_civitai_sha256(dl_url: str, api_key=""):
         return k in qs.keys() and qs[k][0] != json.get(k, None) and json.get(k, None) is not None
 
     if "https://civitai.com/api/download/models/" not in dl_url: return None
-    base_url = 'https://civitai.com/api/v1/model-versions/'
+    base_path = '/model-versions'
     params = {}
     session = create_retry_session(total=6, backoff_factor=1.0)
     m = re.match(r'https://civitai.com/api/download/models/(\d+)\??(.+)?', dl_url)
     if m is None: return None
-    url = base_url + m.group(1)
+    url = f"{base_path}/{m.group(1)}"
     qs = urllib.parse.parse_qs(m.group(2))
     if "type" not in qs.keys(): qs["type"] = ["Model"]
     try:
-        r = civitai_get(session, url, api_key=api_key, params=params, timeout=(5.0, 15), label='Civitai sha256')
+        r = request_civitai_api(session, url, api_key=api_key, params=params, timeout=(5.0, 15), label='Civitai sha256')
         if not r.ok: return None
-        json = dict(r.json())
+        json = dict(get_civitai_response_json(r, default={}) or {})
         if "files" not in json.keys() or not isinstance(json["files"], list): return None
         hash = None
         for d in json["files"]:
@@ -1081,7 +1335,7 @@ def sort_civitai_basemodels(items: list[str]):
 
 
 def fetch_civitai_basemodels(api_key: str = "", pages_per_sort: int = CIVITAI_BASEMODEL_REFRESH_PAGES_PER_SORT):
-    base_url = 'https://civitai.com/api/v1/models'
+    base_path = '/models'
     observed = set()
     session = create_retry_session(total=6, backoff_factor=1.0)
     seeds = [("Newest", "AllTime"), ("Most Downloaded", "AllTime")]
@@ -1091,13 +1345,13 @@ def fetch_civitai_basemodels(api_key: str = "", pages_per_sort: int = CIVITAI_BA
             params = {'sort': sort, 'period': period, 'limit': 100, 'page': page_index, 'nsfw': 'true'}
             try:
                 if next_url is None:
-                    r = civitai_get(session, base_url, api_key=api_key, params=params, timeout=(7.0, 30), label=f'Civitai base model refresh sort={sort} page={page_index}')
+                    r = request_civitai_api(session, base_path, api_key=api_key, params=params, timeout=(7.0, 30), label=f'Civitai base model refresh sort={sort} page={page_index}', source='base-models')
                 else:
-                    r = civitai_get(session, next_url, api_key=api_key, timeout=(7.0, 30), label=f'Civitai base model refresh sort={sort} page={page_index}')
+                    r = request_civitai_api_url(session, next_url, api_key=api_key, timeout=(7.0, 30), label=f'Civitai base model refresh sort={sort} page={page_index}', source='base-models')
                 if not r.ok:
                     print(f"Failed to refresh Civitai base models. sort={sort} page={page_index} status={r.status_code}")
                     break
-                data = r.json()
+                data = get_civitai_response_json(r, default={}) or {}
                 items = data.get('items', [])
                 if not isinstance(items, list): break
                 for item in items:
@@ -1380,8 +1634,6 @@ def begin_probe_feedback(title: str):
     return gr.update(value=f"### {title}\n- status: probing...", visible=True)
 def begin_probe_civitai():
     return begin_probe_feedback("Civitai Probe")
-def begin_probe_civitai_green():
-    return begin_probe_feedback("civitai.green Probe")
 def begin_probe_civitai_keys():
     return begin_probe_feedback("Civitai Key Probe")
 def begin_probe_civitai_url():
@@ -1509,7 +1761,7 @@ def load_all_civitai(state: dict, api_key: str = ""):
 def search_on_civitai(query: str, types: list[str], allow_model: list[str] = [], limit: int = 100,
                       sort: str = "Highest Rated", period: str = "AllTime", tag: str = "", user: str = "", page: int = 1,
                       filetype: list[str] = [], api_key: str = "", base_origin: str = CIVITAI_DEFAULT_ORIGIN, progress=gr.Progress(track_tqdm=False)):
-    base_url = f'{base_origin}/api/v1/models'
+    base_path = '/models'
     api_sort = sort if sort in {"Highest Rated", "Most Downloaded", "Newest"} else CIVITAI_SORT[0]
     params = {'sort': api_sort, 'period': period, 'limit': int(limit), 'nsfw': 'true'}
     clean_types = [str(t).strip() for t in (types or []) if str(t).strip()]
@@ -1530,19 +1782,19 @@ def search_on_civitai(query: str, types: list[str], allow_model: list[str] = [],
         if page == 0:
             progress(0, desc="Searching page 1...")
             print("Searching page 1...")
-            r = civitai_get(session, base_url, api_key=api_key, params=params | {'page': 1}, timeout=(7.0, 30), label='Civitai search page=1', source='search')
+            r = request_civitai_api(session, base_path, api_key=api_key, params=params | {'page': 1}, timeout=(7.0, 30), label='Civitai search page=1', source='search')
             rs.append(r)
             if r.ok:
-                json = r.json()
+                json = get_civitai_response_json(r, default={}) or {}
                 next_url = json['metadata']['nextPage'] if 'metadata' in json and 'nextPage' in json['metadata'] else None
                 i = 2
                 while next_url is not None:
                     progress(0, desc=f"Searching page {i}...")
                     print(f"Searching page {i}...")
-                    r = civitai_get(session, next_url, api_key=api_key, timeout=(7.0, 30), label=f'Civitai search page={i}', source='search')
+                    r = request_civitai_api_url(session, next_url, api_key=api_key, timeout=(7.0, 30), label=f'Civitai search page={i}', source='search')
                     rs.append(r)
                     if r.ok:
-                        json = r.json()
+                        json = get_civitai_response_json(r, default={}) or {}
                         next_url = json['metadata']['nextPage'] if 'metadata' in json and 'nextPage' in json['metadata'] else None
                     else:
                         next_url = None
@@ -1552,18 +1804,18 @@ def search_on_civitai(query: str, types: list[str], allow_model: list[str] = [],
         else:
             progress(0, desc="Searching page 1...")
             print("Searching page 1...")
-            r = civitai_get(session, base_url, api_key=api_key, params=params, timeout=(7.0, 30), label='Civitai search page=1', source='search')
+            r = request_civitai_api(session, base_path, api_key=api_key, params=params, timeout=(7.0, 30), label='Civitai search page=1', source='search')
             rs.append(r)
     except requests.exceptions.ConnectTimeout:
         print("Request timed out.")
     except Exception as e:
         print(e)
     items = []
-    origin = base_origin.rstrip('/')
+    origin = get_civitai_display_origin().rstrip('/')
     for r in rs:
         if not r.ok:
             continue
-        json = r.json()
+        json = get_civitai_response_json(r, default={}) or {}
         if 'items' not in json:
             continue
         for j in json['items']:
@@ -1586,9 +1838,7 @@ def search_on_civitai(query: str, types: list[str], allow_model: list[str] = [],
                     base_item['img_url'] = images[0].get('url', '') or NULL_IMAGE_PATH
                 else:
                     base_item['img_url'] = NULL_IMAGE_PATH
-                model_url = f"{origin}/models/{j.get('id', '')}"
-                if model.get('id') is not None:
-                    model_url += f"?modelVersionId={model.get('id')}"
+                model_url = build_civitai_model_url(j.get('id', ''), model.get('id'))
                 desc = str(base_item.get('description', '') or '')
                 base_item['md'] = (
                     f"Model URL: [{model_url}]({model_url})<br>"
@@ -1603,7 +1853,7 @@ def search_on_civitai(query: str, types: list[str], allow_model: list[str] = [],
                 if files:
                     for f in files:
                         item = base_item.copy()
-                        item['dl_url'] = f.get('downloadUrl', '')
+                        item['dl_url'] = normalize_civitai_download_api_url(f.get('downloadUrl', ''))
                         item['size_kb'] = f.get('sizeKB', 0.0)
                         item['file_type'] = f.get('type', '')
                         if len(filetype) != 0 and f.get('type', '') not in set(filetype):
@@ -1611,7 +1861,7 @@ def search_on_civitai(query: str, types: list[str], allow_model: list[str] = [],
                         items.append(item)
                 else:
                     item = base_item.copy()
-                    item['dl_url'] = model.get('downloadUrl', '')
+                    item['dl_url'] = normalize_civitai_download_api_url(model.get('downloadUrl', ''))
                     items.append(item)
     if sort in CIVITAI_SORT_EXT:
         if sort == "Size":
@@ -1686,17 +1936,17 @@ def get_civitai_json(dl_url: str, is_html: bool=False, image_baseurl: str="", ap
     default = ("", "", "") if is_html else ""
     if "https://civitai.com/api/download/models/" not in dl_url:
         return default
-    base_url = 'https://civitai.com/api/v1/model-versions/'
+    base_path = '/model-versions'
     params = {}
     session = create_retry_session(total=6, backoff_factor=1.0)
     model_id = re.sub(r'https://civitai.com/api/download/models/(\d+)(?:.+)?', r'\1', dl_url)
-    url = base_url + model_id
+    url = f"{base_path}/{model_id}"
     try:
         reset_civitai_key_status(api_key, source='info')
-        r = civitai_get(session, url, api_key=api_key, params=params, timeout=(5.0, 60), label='Civitai model-version', source='info')
+        r = request_civitai_api(session, url, api_key=api_key, params=params, timeout=(5.0, 60), label='Civitai model-version', source='info')
         if not r.ok:
             return default
-        json = dict(r.json()).copy()
+        json = dict(get_civitai_response_json(r, default={}) or {}).copy()
         html = ""
         image = ""
         if "modelId" in json.keys():
@@ -1704,7 +1954,7 @@ def get_civitai_json(dl_url: str, is_html: bool=False, image_baseurl: str="", ap
             if is_civitai_host(original_parts.netloc) and re.match(r'^/models/\d+(?:/[^/?#]+)?/?$', original_parts.path or ""):
                 url = original_dl_url
             else:
-                url = f"https://civitai.com/models/{json['modelId']}"
+                url = build_civitai_model_url(json['modelId'])
             r = civitai_get(session, url, api_key=api_key, params=params, timeout=(5.0, 60), label='Civitai model page', source='info')
             if not r.ok:
                 return json, html, image
@@ -1742,12 +1992,13 @@ def get_civitai_json(dl_url: str, is_html: bool=False, image_baseurl: str="", ap
         return default
 
 
-def _load_civitai_choice_list(base_url: str, label: str, source: str, value_key: str, count_key: str,
+def _load_civitai_choice_list(api_path: str, label: str, source: str, value_key: str, count_key: str,
                              query: str="", api_key: str="", limit: int=200,
                              timeout: tuple[float, float]=(7.0, 15.0), retry_total: int=6,
                              backoff_factor: float=1.0, default: list[str] | None=None,
                              cache: dict | None=None, cache_lock: object | None=None,
-                             cache_ttl_sec: int=0, startup_name: str=""):
+                             cache_ttl_sec: int=0, startup_name: str="", preferred_origin: str="",
+                             non_json_fallback_origin: str=""):
     clean_query = str(query or "").strip()
     cache_key = clean_query.casefold()
     now = time.time()
@@ -1762,12 +2013,22 @@ def _load_civitai_choice_list(base_url: str, label: str, source: str, value_key:
         params['query'] = clean_query
     session = create_retry_session(total=retry_total, backoff_factor=backoff_factor)
     try:
-        r = civitai_get(session, base_url, api_key=api_key, params=params, timeout=timeout, label=label, source=source)
+        r = request_civitai_api(
+            session,
+            api_path,
+            api_key=api_key,
+            params=params,
+            timeout=timeout,
+            label=label,
+            source=source,
+            preferred_origin=preferred_origin,
+            non_json_fallback_origin=non_json_fallback_origin,
+        )
         if not r.ok:
             if not clean_query and startup_name:
                 print(f"Failed to load {startup_name} from API at startup. HTTP {r.status_code}.")
             return default
-        j = dict(r.json()).copy()
+        j = dict(get_civitai_response_json(r, default={}) or {}).copy()
         items = j.get('items', []) if isinstance(j.get('items', []), list) else []
         observed = []
         for item in items:
@@ -1790,7 +2051,7 @@ def _load_civitai_choice_list(base_url: str, label: str, source: str, value_key:
 
 def get_civitai_creator(query: str="", api_key: str="", limit: int=CREATOR_SUGGEST_LIMIT):
     return _load_civitai_choice_list(
-        base_url='https://civitai.com/api/v1/creators',
+        api_path='/creators',
         label='Civitai creators',
         source='creators',
         value_key='username',
@@ -1806,6 +2067,7 @@ def get_civitai_creator(query: str="", api_key: str="", limit: int=CREATOR_SUGGE
         cache_lock=CREATOR_SUGGEST_LOCK,
         cache_ttl_sec=CREATOR_CACHE_TTL_SEC,
         startup_name='Civitai creators',
+        non_json_fallback_origin=CIVITAI_DEFAULT_ORIGIN,
     )
 
 
@@ -1827,7 +2089,7 @@ def refresh_civitai_creators(user_value: str="", api_key: str=""):
 
 def get_civitai_tag():
     return _load_civitai_choice_list(
-        base_url='https://civitai.com/api/v1/tags',
+        api_path='/tags',
         label='Civitai tags',
         source='tags',
         value_key='name',
@@ -1961,6 +2223,60 @@ def extract_probe_first_download_url(payload: dict):
                 return dl_url, model, version, {}
     return "", {}, {}, {}
 
+def probe_json_api_step(session, path: str, *, api_key: str = "", params=None, timeout: tuple[float, float] = (7.0, 20.0),
+                        label: str = "Civitai probe", source: str = "probe", preferred_origin: str = "",
+                        non_json_fallback_origin: str = ""):
+    response = None
+    try:
+        response = request_civitai_api(
+            session,
+            path,
+            api_key=api_key,
+            params=params,
+            timeout=timeout,
+            label=label,
+            source=source,
+            preferred_origin=preferred_origin,
+            non_json_fallback_origin=non_json_fallback_origin,
+        )
+        ok = bool(response is not None and getattr(response, "ok", False))
+        status = str(getattr(response, "status_code", "-"))
+        response_url = str(getattr(response, "url", "") or "")
+        response_host = urllib.parse.urlparse(response_url).netloc
+        json_ok = False
+        if ok:
+            payload = get_civitai_response_json(response, default=_CIVITAI_JSON_MISSING)
+            json_ok = payload is not _CIVITAI_JSON_MISSING
+        return {"ok": ok, "status": status, "host": response_host, "json_ok": json_ok, "error": ""}
+    except Exception as e:
+        return {"ok": False, "status": "-", "host": "", "json_ok": False, "error": f"{type(e).__name__}: {e}"}
+    finally:
+        try:
+            if response is not None:
+                response.close()
+        except Exception:
+            pass
+
+
+def probe_download_url_step(session, download_url: str, *, api_key: str = "", timeout: tuple[float, float] = (7.0, 20.0)):
+    response = None
+    try:
+        response = civitai_get(session, download_url, api_key=api_key, timeout=timeout, label='Civitai probe download', source='probe-download')
+        status = str(getattr(response, "status_code", "-"))
+        status_code = int(status) if str(status).isdigit() else 0
+        response_url = str(getattr(response, "url", "") or download_url or "")
+        response_host = urllib.parse.urlparse(response_url).netloc
+        ok = bool(response is not None and (getattr(response, "ok", False) or status_code in {301, 302, 303, 307, 308}))
+        return {"ok": ok, "status": status, "host": response_host, "error": ""}
+    except Exception as e:
+        return {"ok": False, "status": "-", "host": "", "error": f"{type(e).__name__}: {e}"}
+    finally:
+        try:
+            if response is not None:
+                response.close()
+        except Exception:
+            pass
+
 
 def probe_civitai_keys(api_key: str):
     effective_api_key = get_effective_probe_api_key(api_key)
@@ -1975,10 +2291,12 @@ def probe_civitai_keys(api_key: str):
         return gr.update(value=probe_text, visible=True), gr.update(value=format_civitai_key_status_md(effective_api_key))
     session = create_retry_session(total=4, backoff_factor=0.8)
     try:
+        api_origin = resolve_civitai_api_origin(session)
+        lines.append(f"- api origin: {api_origin}")
         reset_civitai_key_status(effective_api_key, source="probe-keys")
-        resp = civitai_get(
+        resp = request_civitai_api(
             session,
-            f"{CIVITAI_DEFAULT_ORIGIN}/api/v1/models",
+            "/models",
             api_key=effective_api_key,
             params={"limit": 1, "sort": "Newest", "period": "AllTime", "nsfw": "true"},
             timeout=(7.0, 20.0),
@@ -2137,63 +2455,130 @@ def probe_civitai_sidecar(url_input: str, fallback_query: str, api_key: str):
     return gr.update(value=probe_text, visible=True), gr.update(value=format_civitai_key_status_md(effective_api_key))
 
 def probe_civitai_api(query: str, api_key: str):
-    origin = CIVITAI_DEFAULT_ORIGIN
+    page_origin = get_civitai_display_origin()
     effective_api_key = get_effective_probe_api_key(api_key)
     key_source = "input" if str(api_key or "").strip() else ("env" if effective_api_key else "none")
     reset_civitai_key_status(effective_api_key, source="probe")
     probe_query = str(query or "lora").strip() or "lora"
     print(f"Civitai Probe: query={probe_query!r} key_source={key_source}")
-    lines = ["### Civitai Probe", f"- origin: {origin}", f"- query: {probe_query}", f"- key source: {key_source}"]
+    lines = ["### Civitai Probe", f"- page origin: {page_origin}", f"- query: {probe_query}", f"- key source: {key_source}"]
     session = create_retry_session(total=4, backoff_factor=0.8)
+    api_origin = resolve_civitai_api_origin(session)
+    lines.insert(1, f"- api origin: {api_origin}")
     anon_ok = False
     try:
-        page_resp = session.get(f"{origin}/models", params={"query": probe_query}, headers=get_civitai_headers(""), timeout=(7.0, 20.0))
+        page_resp = session.get(f"{page_origin}/models", params={"query": probe_query}, headers=get_civitai_headers(""), timeout=(7.0, 20.0))
         lines.append(f"- anonymous page: {'ok' if page_resp.ok else 'fail'} ({page_resp.status_code})")
         if page_resp.ok:
             first_path = extract_first_model_path(page_resp.text)
             if first_path:
                 lines.append(f"- anonymous first model: {first_path}")
-        anon_resp = session.get(
-            f"{origin}/api/v1/models",
+        anon_resp = request_civitai_api(
+            session,
+            "/models",
             params={"query": probe_query, "limit": 1, "sort": "Newest", "period": "AllTime", "nsfw": "true"},
-            headers=get_civitai_headers(""),
             timeout=(7.0, 20.0),
+            label='Civitai anonymous probe models',
+            source='probe-anon',
         )
-        lines.append(f"- anonymous api: {'ok' if anon_resp.ok else 'fail'} ({anon_resp.status_code})")
-        anon_ok = bool(anon_resp.ok)
+        anon_status = getattr(anon_resp, "status_code", "-")
+        lines.append(f"- anonymous api: {'ok' if anon_resp and anon_resp.ok else 'fail'} ({anon_status})")
+        anon_ok = bool(anon_resp is not None and anon_resp.ok)
     except Exception as e:
         lines.append(f"- anonymous probe: fail ({type(e).__name__}: {e})")
+
+    tags_probe = probe_json_api_step(
+        session,
+        "/tags",
+        params={"limit": 1},
+        timeout=(7.0, 20.0),
+        label='Civitai probe tags',
+        source='probe-tags',
+    )
+    if tags_probe["error"]:
+        lines.append(f"- tags api: fail ({tags_probe['error']})")
+    else:
+        tags_state = 'ok' if tags_probe["ok"] and tags_probe["json_ok"] else 'fail'
+        lines.append(f"- tags api: {tags_state} ({tags_probe['status']}, {tags_probe['host'] or '-'})")
+
+    creators_probe = probe_json_api_step(
+        session,
+        "/creators",
+        api_key=effective_api_key,
+        params={"limit": 1, "query": probe_query},
+        timeout=(7.0, 20.0),
+        label='Civitai probe creators',
+        source='probe-creators',
+        preferred_origin=CIVITAI_RED_ORIGIN,
+        non_json_fallback_origin=CIVITAI_DEFAULT_ORIGIN,
+    )
+    if creators_probe["error"]:
+        lines.append(f"- creators api: fail ({creators_probe['error']})")
+    else:
+        creators_state = 'ok' if creators_probe["ok"] and creators_probe["json_ok"] else 'fail'
+        creator_host = canonicalize_civitai_host(creators_probe["host"]) if creators_probe["host"] else ''
+        lines.append(f"- creators api: {creators_state} ({creators_probe['status']}, {creators_probe['host'] or '-'})")
+        if creator_host == 'civitai.com':
+            lines.append("- creators fallback: ok (civitai.com)")
+        elif creator_host == 'civitai.red':
+            lines.append("- creators fallback: not-needed (civitai.red json)")
+
     parsed_keys = parse_civitai_api_keys(effective_api_key)
     if not parsed_keys:
         lines.append("- auth resolve: skipped (no Civitai key)")
+        lines.append("- direct download: skipped (no Civitai key)")
         probe_text = "\n".join(lines)
         print(probe_text)
         return gr.update(value=probe_text, visible=True), gr.update(value=format_civitai_key_status_md(effective_api_key))
+
     try:
         reset_civitai_key_status(effective_api_key, source="probe-auth")
-        auth_resp = civitai_get(
+        auth_resp = request_civitai_api(
             session,
-            f"{origin}/api/v1/models",
+            "/models",
             api_key=effective_api_key,
             params={"query": probe_query, "limit": 1, "sort": "Newest", "period": "AllTime", "nsfw": "true"},
             timeout=(7.0, 20.0),
             label='Civitai probe models',
             source='probe-auth',
         )
-        lines.append(f"- auth api: {'ok' if auth_resp.ok else 'fail'} ({auth_resp.status_code})")
-        if auth_resp.ok:
-            payload = auth_resp.json() if auth_resp.content else {}
+        auth_status = getattr(auth_resp, "status_code", "-")
+        lines.append(f"- auth api: {'ok' if auth_resp and auth_resp.ok else 'fail'} ({auth_status})")
+        if auth_resp is not None and auth_resp.ok:
+            payload = get_civitai_response_json(auth_resp, default={}) if auth_resp.content else {}
             dl_url, model, version, file_info = extract_probe_first_download_url(payload)
             if dl_url:
                 lines.append(f"- auth first file: {model.get('name', '')} / {version.get('name', '')} / {round(float(file_info.get('sizeKB', 0.0) or 0.0) / 1000.0, 2)}MB")
+                model_url = build_civitai_model_url(model.get('id'), version.get('id'))
+                normalized_model_download = normalize_civitai_input_url(model_url, api_key=effective_api_key)
+                normalized_auth_download = normalize_civitai_download_api_url(dl_url)
+                model_download_parts = get_civitai_url_parts(normalized_model_download)
+                auth_download_parts = get_civitai_url_parts(normalized_auth_download)
+                same_download_path = bool(
+                    is_civitai_download_api_path(model_download_parts.path)
+                    and is_civitai_download_api_path(auth_download_parts.path)
+                    and model_download_parts.path == auth_download_parts.path
+                )
+                lines.append(
+                    f"- model page normalize: {'ok' if same_download_path else 'mismatch'} ({sanitize_url_for_log(normalized_model_download)})"
+                )
                 try:
-                    resolved_url = resolve_civitai_download_url(dl_url, effective_api_key, max_tries=1)
+                    resolved_url = resolve_civitai_download_url(normalized_auth_download, effective_api_key, max_tries=1)
                     resolved_host = urllib.parse.urlparse(resolved_url).netloc
                     lines.append(f"- auth resolve: ok ({resolved_host})")
+                    download_probe = probe_download_url_step(session, resolved_url, api_key=effective_api_key, timeout=(7.0, 20.0))
+                    if download_probe["error"]:
+                        lines.append(f"- direct download: fail ({download_probe['error']})")
+                    else:
+                        lines.append(
+                            f"- direct download: {'ok' if download_probe['ok'] else 'fail'} ({download_probe['status']}, {download_probe['host'] or '-'})"
+                        )
                 except Exception as e:
                     lines.append(f"- auth resolve: fail ({type(e).__name__}: {e})")
+                    lines.append("- direct download: skipped (resolve failed)")
             else:
                 lines.append("- auth resolve: skipped (no downloadUrl in first result)")
+                lines.append("- direct download: skipped (no downloadUrl in first result)")
     except Exception as e:
         lines.append(f"- auth probe: fail ({type(e).__name__}: {e})")
     if not anon_ok:
@@ -2201,51 +2586,3 @@ def probe_civitai_api(query: str, api_key: str):
     probe_text = "\n".join(lines)
     print(probe_text)
     return gr.update(value=probe_text, visible=True), gr.update(value=format_civitai_key_status_md(effective_api_key))
-def probe_civitai_green_api(query: str, api_key: str):
-    effective_api_key = get_effective_probe_api_key(api_key)
-    key_source = "input" if str(api_key or "").strip() else ("env" if effective_api_key else "none")
-    probe_input = str(query or "").strip() or "lora"
-    print(f"civitai.green Probe: input={probe_input!r} key_source={key_source}")
-    lines = ["### civitai.green Probe", f"- input: {probe_input}", f"- key source: {key_source}"]
-    session = create_retry_session(total=4, backoff_factor=0.8)
-    try:
-        parts = get_civitai_url_parts(probe_input)
-        if is_civitai_host(parts.netloc):
-            lines.append(f"- url host: {parts.netloc}")
-            normalized = normalize_civitai_input_url(probe_input, api_key=effective_api_key)
-            if normalized != probe_input:
-                lines.append(f"- normalized download url: {normalize_civitai_download_api_url(normalized)}")
-            if "api/download/models/" in normalized:
-                lines.append("- mode: direct-download probe")
-                try:
-                    resolved_url = resolve_civitai_download_url(normalized, effective_api_key, max_tries=1)
-                    resolved_host = urllib.parse.urlparse(resolved_url).netloc
-                    lines.append(f"- resolve: ok ({resolved_host})")
-                except Exception as e:
-                    lines.append(f"- resolve: fail ({type(e).__name__}: {e})")
-            else:
-                lines.append("- mode: page-only probe")
-                page_resp = session.get(probe_input, headers=get_civitai_headers(effective_api_key), timeout=(7.0, 20.0))
-                lines.append(f"- page: {'ok' if page_resp.ok else 'fail'} ({page_resp.status_code})")
-                if page_resp.ok:
-                    first_path = extract_first_model_path(page_resp.text)
-                    if first_path:
-                        lines.append(f"- first model: {first_path}")
-                    first_download = extract_first_civitai_download_url_from_html(page_resp.text)
-                    lines.append(f"- embedded download url: {'found' if first_download else 'not found'}")
-        else:
-            origin = CIVITAI_GREEN_ORIGIN
-            lines.append(f"- origin: {origin}")
-            lines.append("- mode: query probe")
-            page_resp = session.get(f"{origin}/models", params={"query": probe_input}, headers=get_civitai_headers(effective_api_key), timeout=(7.0, 20.0))
-            lines.append(f"- page: {'ok' if page_resp.ok else 'fail'} ({page_resp.status_code})")
-            if page_resp.ok:
-                first_path = extract_first_model_path(page_resp.text)
-                if first_path:
-                    lines.append(f"- first model: {first_path}")
-            lines.append("- api: skipped (HTML-only probe)")
-    except Exception as e:
-        lines.append(f"- probe: fail ({type(e).__name__}: {e})")
-    probe_text = "\n".join(lines)
-    print(probe_text)
-    return gr.update(value=probe_text, visible=True), gr.update(value=format_civitai_key_status_md(effective_api_key))