Upload modutils.py

modutils.py

@@ -22,6 +22,8 @@ FILENAME_TIMEZONE = timezone(timedelta(hours=9)) # JST
 import torch
 from safetensors.torch import load_file
 import gc
+import html as html_lib
+import subprocess
 
 
 from env import (HF_LORA_PRIVATE_REPOS1, HF_LORA_PRIVATE_REPOS2,
@@ -139,44 +141,431 @@ def download_hf_file(directory, url, force_filename="", hf_token="", progress=gr
 
 
 USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.0'
+CIVITAI_DEFAULT_ORIGIN = "https://civitai.com"
+CIVITAI_REFERER = f"{CIVITAI_DEFAULT_ORIGIN}/"
+CIVITAI_HOST_ALIASES = frozenset({"civitai.com", "www.civitai.com", "civitai.green", "www.civitai.green"})
+CIVITAI_RETRY_TOTAL = 5
+CIVITAI_RETRY_BACKOFF = 1.0
+CIVITAI_RESOLVE_RETRY_TOTAL = 4
+CIVITAI_RESOLVE_RETRY_BACKOFF = 0.8
+CIVITAI_STATUS_FORCELIST = [429, 500, 502, 503, 504]
+CIVITAI_RESOLVE_TIMEOUT = (7.0, 25.0)
+CIVITAI_METADATA_TIMEOUT = (3.0, 15.0)
+CIVITAI_SEARCH_TIMEOUT = (3.0, 30.0)
+CIVITAI_NEGATIVE_CACHE_LIMIT = 256
+CIVITAI_RESOLVE_CACHE: dict[str, str] = {}
+CIVITAI_RESOLVE_NEGATIVE_CACHE: dict[str, str] = {}
+CIVITAI_VERSION_JSON_CACHE: dict[str, dict] = {}
+CIVITAI_VERSION_NEGATIVE_CACHE: dict[str, str] = {}
+CIVITAI_ARIA2_CONNECTIONS = 1
+CIVITAI_ARIA2_SPLIT = 1
+CIVITAI_ARIA2_MIN_SPLIT_SIZE = "1M"
+CIVITAI_ARIA2_RESUME = False
+CIVITAI_ARIA2_FRESH_RETRY_LIMIT = 1
+
+
+def create_retry_session(total=CIVITAI_RETRY_TOTAL, backoff_factor=CIVITAI_RETRY_BACKOFF):
+    session = requests.Session()
+    retries = Retry(total=total, backoff_factor=backoff_factor, status_forcelist=CIVITAI_STATUS_FORCELIST)
+    session.mount("https://", HTTPAdapter(max_retries=retries))
+    session.mount("http://", HTTPAdapter(max_retries=retries))
+    return session
 
 
-def request_json_data(url):
-    model_version_id = url.split('/')[-1]
-    if "?modelVersionId=" in model_version_id:
-        match = re.search(r'modelVersionId=(\d+)', url)
-        model_version_id = match.group(1)
+def cache_put(cache: dict, key: str, value):
+    key = str(key or "").strip()
+    if not key:
+        return
+    if key in cache:
+        cache.pop(key, None)
+    elif len(cache) >= CIVITAI_NEGATIVE_CACHE_LIMIT:
+        try:
+            cache.pop(next(iter(cache)))
+        except Exception:
+            cache.clear()
+    cache[key] = value
 
-    endpoint_url = f"https://civitai.com/api/v1/model-versions/{model_version_id}"
 
-    params = {}
+def get_civitai_headers(api_key: str = ""):
     headers = {'User-Agent': USER_AGENT, 'content-type': 'application/json'}
-    session = requests.Session()
-    retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
-    session.mount("https://", HTTPAdapter(max_retries=retries))
+    if api_key:
+        headers['Authorization'] = f'Bearer {api_key}'
+    return headers
+
+
+def get_civitai_url_parts(url: str):
+    try:
+        return urllib.parse.urlsplit(str(url or "").strip())
+    except Exception:
+        return urllib.parse.urlsplit("")
+
+
+def sanitize_url_for_log(url: str):
+    raw = str(url or "").strip()
+    if not raw:
+        return raw
+    parts = get_civitai_url_parts(raw)
+    if not parts.netloc:
+        return raw
+    pairs = [
+        (k, v)
+        for k, v in urllib.parse.parse_qsl(parts.query, keep_blank_values=True)
+        if str(k).lower() != "token"
+    ]
+    query = urllib.parse.urlencode(pairs)
+    return urllib.parse.urlunsplit((parts.scheme, parts.netloc, parts.path, query, parts.fragment))
+
+
+def is_civitai_host(netloc: str):
+    return str(netloc or "").strip().lower() in CIVITAI_HOST_ALIASES
+
+
+def is_civitai_url(url: str):
+    return is_civitai_host(get_civitai_url_parts(url).netloc)
+
+
+def is_civitai_download_api_path(path: str):
+    return re.match(r'^/api/download/models/\d+$', str(path or "").strip()) is not None
+
+
+def extract_civitai_model_version_id(url: str):
+    try:
+        parts = get_civitai_url_parts(url)
+        for pattern in [r'^/api/download/models/(\d+)$', r'^/api/v1/model-versions/(\d+)$']:
+            m = re.match(pattern, str(parts.path or "").strip())
+            if m:
+                return m.group(1)
+        qs = urllib.parse.parse_qs(parts.query)
+        for key in ["modelVersionId", "modelversionid", "versionId", "versionid"]:
+            values = qs.get(key, [])
+            if not values:
+                continue
+            value = str(values[0]).strip()
+            if value.isdigit():
+                return value
+    except Exception:
+        return ""
+    return ""
+
+
+def get_civitai_query_filters(url: str):
+    try:
+        parts = get_civitai_url_parts(url)
+        qs = urllib.parse.parse_qs(parts.query)
+    except Exception:
+        return {}
+    filters = {}
+    for key in ["type", "format", "size", "fp"]:
+        values = qs.get(key, [])
+        if values:
+            filters[key] = str(values[0]).strip()
+    return filters
+
+
+def normalize_civitai_filter_value(key: str, value):
+    if value is None:
+        return ""
+    text = str(value).strip()
+    if not text:
+        return ""
+    if key == "fp":
+        return text.replace("-", "").replace("_", "").replace(" ", "").lower()
+    return text.lower()
 
+
+def describe_civitai_file_for_log(file_info):
+    if not isinstance(file_info, dict):
+        return ""
+    parts = []
+    for key in ["name", "type", "format", "size", "fp"]:
+        value = file_info.get(key)
+        if value is None:
+            continue
+        text = str(value).strip()
+        if text:
+            parts.append(f"{key}={text}")
+    hashes = file_info.get("hashes") if isinstance(file_info.get("hashes"), dict) else {}
+    sha256 = str(hashes.get("SHA256") or "").strip()
+    if sha256:
+        parts.append(f"sha256={sha256[:12]}...")
+    return ", ".join(parts)
+
+
+def build_civitai_download_query_from_url(url: str):
     try:
-        result = session.get(endpoint_url, params=params, headers=headers, stream=True, timeout=(3.0, 15))
+        parts = get_civitai_url_parts(url)
+    except Exception:
+        return ""
+    blocked = {"modelversionid", "versionid"}
+    pairs = [
+        (k, v)
+        for k, v in urllib.parse.parse_qsl(parts.query, keep_blank_values=True)
+        if str(k).lower() not in blocked
+    ]
+    return urllib.parse.urlencode(pairs)
+
+
+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}"
+    return f"{base}?{query}" if query else base
+
+
+def normalize_civitai_download_api_url(url: str):
+    parts = get_civitai_url_parts(url)
+    if not is_civitai_host(parts.netloc) or not is_civitai_download_api_path(parts.path):
+        return str(url or "").strip()
+    return urllib.parse.urlunsplit(("https", "civitai.com", parts.path, parts.query, ""))
+
+
+def extract_first_civitai_download_url_from_html(html: str):
+    if not html:
+        return ""
+    page = html_lib.unescape(str(html))
+    patterns = [
+        r'https?://(?:www\.)?(?:civitai\.com|civitai\.green)/api/download/models/\d+[^\s\'\"<>\)\]\}]*',
+        r'["\'](/api/download/models/\d+[^"\']*)["\']',
+    ]
+    for pattern in patterns:
+        try:
+            m = re.search(pattern, page, flags=re.IGNORECASE)
+        except re.error:
+            m = None
+        if not m:
+            continue
+        candidate = m.group(1) if m.lastindex else m.group(0)
+        candidate = str(candidate or "").strip("\"'")
+        if candidate.startswith("/"):
+            candidate = urllib.parse.urljoin(CIVITAI_DEFAULT_ORIGIN, candidate)
+        return normalize_civitai_download_api_url(candidate)
+    return ""
+
+
+def resolve_civitai_model_page_to_download_url(url: str, api_key: str = ""):
+    raw = str(url or "").strip()
+    if not raw:
+        return raw
+    cached = CIVITAI_RESOLVE_CACHE.get(raw)
+    if cached:
+        return cached
+    if raw in CIVITAI_RESOLVE_NEGATIVE_CACHE:
+        return raw
+    parts = get_civitai_url_parts(raw)
+    if not is_civitai_host(parts.netloc):
+        return raw
+    if is_civitai_download_api_path(parts.path):
+        normalized = normalize_civitai_download_api_url(raw)
+        cache_put(CIVITAI_RESOLVE_CACHE, raw, normalized)
+        return normalized
+    if not re.match(r'^/models/\d+(?:/[^/?#]+)?/?$', parts.path or ""):
+        return raw
+    version_id = extract_civitai_model_version_id(raw)
+    if version_id:
+        normalized = to_civitai_default_download_url(version_id, query=build_civitai_download_query_from_url(raw))
+        cache_put(CIVITAI_RESOLVE_CACHE, raw, normalized)
+        return normalized
+    headers = get_civitai_headers(api_key if parts.netloc.lower().endswith("civitai.com") else "")
+    headers['Referer'] = f"{parts.scheme or 'https'}://{parts.netloc}/"
+    session = create_retry_session(total=CIVITAI_RESOLVE_RETRY_TOTAL, backoff_factor=CIVITAI_RESOLVE_RETRY_BACKOFF)
+    try:
+        r = session.get(raw, headers=headers, timeout=CIVITAI_RESOLVE_TIMEOUT)
+        if not r.ok:
+            print(f"Civitai model page resolve failed: {sanitize_url_for_log(raw)} status={r.status_code}")
+            if r.status_code in [400, 401, 403, 404]:
+                cache_put(CIVITAI_RESOLVE_NEGATIVE_CACHE, raw, f"status={r.status_code}")
+            return raw
+        extracted = extract_first_civitai_download_url_from_html(r.text)
+        if extracted:
+            normalized = normalize_civitai_download_api_url(extracted)
+            cache_put(CIVITAI_RESOLVE_CACHE, raw, normalized)
+            return normalized
+        cache_put(CIVITAI_RESOLVE_NEGATIVE_CACHE, raw, "html_no_download_url")
+        return raw
+    except Exception as e:
+        print(f"Failed to resolve Civitai model page URL: {sanitize_url_for_log(raw)} {type(e).__name__}: {e}")
+        return raw
+
+
+def normalize_civitai_input_url(url: str, api_key: str = ""):
+    raw = str(url or "").strip()
+    if not raw or not is_civitai_url(raw):
+        return raw
+    normalized = resolve_civitai_model_page_to_download_url(raw, api_key=api_key)
+    if normalized != raw:
+        print(f"Normalized Civitai URL: {sanitize_url_for_log(raw)} -> {sanitize_url_for_log(normalized)}")
+    return normalized
+
+
+def append_civitai_token(url: str, api_key: str = ""):
+    raw = str(url or "").strip()
+    if not raw or not api_key:
+        return raw
+    parts = get_civitai_url_parts(raw)
+    pairs = [(k, v) for k, v in urllib.parse.parse_qsl(parts.query, keep_blank_values=True) if k.lower() != "token"]
+    pairs.append(("token", api_key))
+    query = urllib.parse.urlencode(pairs)
+    return urllib.parse.urlunsplit((parts.scheme or "https", parts.netloc, parts.path, query, parts.fragment))
+
+
+def get_civitai_request_context(url: str, api_key: str = ""):
+    raw_url = str(url or "").strip()
+    normalized_url = normalize_civitai_input_url(raw_url, api_key=api_key)
+    model_version_id = extract_civitai_model_version_id(normalized_url) or extract_civitai_model_version_id(raw_url)
+    return {
+        "raw_url": raw_url,
+        "normalized_url": normalized_url,
+        "model_version_id": model_version_id,
+        "filters": get_civitai_query_filters(raw_url),
+    }
+
+
+def resolve_civitai_download_url(url: str, civitai_api_key: str = "", max_tries: int = 3):
+    raw = normalize_civitai_download_api_url(str(url or "").strip())
+    if not raw:
+        return raw
+    headers = get_civitai_headers(civitai_api_key)
+    headers["Referer"] = CIVITAI_REFERER
+    dl_url = append_civitai_token(raw, civitai_api_key)
+    last_error = None
+    for attempt in range(1, max_tries + 1):
+        response = None
+        try:
+            response = create_retry_session(total=3, backoff_factor=1.0).get(
+                dl_url,
+                headers=headers,
+                allow_redirects=False,
+                stream=True,
+                timeout=CIVITAI_RESOLVE_TIMEOUT,
+            )
+            status = int(response.status_code)
+            location = str(response.headers.get("Location") or "").strip()
+            resolved_url = str(location or response.url or dl_url).strip()
+            resolved_host = get_civitai_url_parts(resolved_url).netloc
+            print(
+                f"[civitai] resolve signed url attempt={attempt}/{max_tries} status={status} "
+                f"host={resolved_host or '-'} url={sanitize_url_for_log(raw)}"
+            )
+            if status in (301, 302, 303, 307, 308) and location:
+                return resolved_url
+            if response.ok and resolved_url and not is_civitai_host(resolved_host):
+                return resolved_url
+            last_error = RuntimeError(f"status={status}")
+        except Exception as e:
+            last_error = e
+            print(
+                f"[civitai] resolve signed url failed attempt={attempt}/{max_tries} "
+                f"url={sanitize_url_for_log(raw)} error={type(e).__name__}: {e}"
+            )
+        finally:
+            try:
+                if response is not None:
+                    response.close()
+            except Exception:
+                pass
+        if attempt < max_tries:
+            time.sleep(min(3.0, 0.8 * attempt))
+    if last_error is not None:
+        raise last_error
+    raise RuntimeError("Failed to resolve Civitai signed download URL")
+
+
+def pick_civitai_file_from_version_json(json_data, source_url: str = ""):
+    files = json_data.get("files", []) if isinstance(json_data, dict) else []
+    if not isinstance(files, list) or not files:
+        return {}
+    version_id = str((json_data or {}).get("id") or "")
+    filters = get_civitai_query_filters(source_url)
+    candidates = []
+    fallback = []
+    for idx, file_info in enumerate(files):
+        if not isinstance(file_info, dict):
+            continue
+        mismatch = False
+        matched_filter_count = 0
+        for key, expected in filters.items():
+            actual = file_info.get(key)
+            expected_norm = normalize_civitai_filter_value(key, expected)
+            actual_norm = normalize_civitai_filter_value(key, actual)
+            if actual_norm:
+                if actual_norm != expected_norm:
+                    mismatch = True
+                    break
+                matched_filter_count += 1
+        download_url = str(file_info.get("downloadUrl") or "")
+        score = 0
+        if matched_filter_count:
+            score += matched_filter_count * 3
+        if version_id and version_id in download_url:
+            score += 4
+        if download_url:
+            score += 2
+        if file_info.get("name"):
+            score += 1
+        hashes = file_info.get("hashes") if isinstance(file_info.get("hashes"), dict) else {}
+        if str(hashes.get("SHA256") or "").strip():
+            score += 1
+        target = fallback if mismatch else candidates
+        target.append((score, idx, file_info))
+    pool = candidates if candidates else fallback
+    if not pool:
+        return {}
+    pool.sort(key=lambda item: (item[0], item[1]), reverse=True)
+    return dict(pool[0][2])
+
+
+def request_json_data(url):
+    context = get_civitai_request_context(url, api_key=CIVITAI_API_KEY)
+    raw_url = context["raw_url"]
+    normalized_url = context["normalized_url"]
+    model_version_id = context["model_version_id"]
+    if not model_version_id:
+        print(f"Civitai metadata lookup skipped: modelVersionId not found for {sanitize_url_for_log(raw_url)}")
+        cache_put(CIVITAI_RESOLVE_NEGATIVE_CACHE, raw_url, "missing_model_version_id")
+        return None
+
+    cached_json = CIVITAI_VERSION_JSON_CACHE.get(model_version_id)
+    if cached_json:
+        return copy.deepcopy(cached_json)
+    if model_version_id in CIVITAI_VERSION_NEGATIVE_CACHE:
+        return None
+
+    endpoint_url = f"https://civitai.com/api/v1/model-versions/{model_version_id}"
+    headers = get_civitai_headers(CIVITAI_API_KEY)
+    session = create_retry_session()
+
+    try:
+        result = session.get(endpoint_url, headers=headers, stream=True, timeout=CIVITAI_METADATA_TIMEOUT)
+        if result.status_code == 404:
+            print(f"Civitai metadata lookup status=404: {endpoint_url}")
+            cache_put(CIVITAI_VERSION_NEGATIVE_CACHE, model_version_id, "status=404")
+            return None
         result.raise_for_status()
         json_data = result.json()
-        return json_data if json_data else None
+        if not json_data:
+            print(f"Civitai metadata lookup returned empty JSON: {endpoint_url}")
+            cache_put(CIVITAI_VERSION_NEGATIVE_CACHE, model_version_id, "empty_json")
+            return None
+        cache_put(CIVITAI_VERSION_JSON_CACHE, model_version_id, copy.deepcopy(json_data))
+        if normalized_url and normalized_url != raw_url:
+            cache_put(CIVITAI_RESOLVE_CACHE, raw_url, normalized_url)
+        return json_data
     except Exception as e:
-        print(f"Error: {e}")
+        print(f"Civitai metadata lookup failed: {endpoint_url} {type(e).__name__}: {e}")
         return None
 
 
 class ModelInformation:
-    def __init__(self, json_data):
+    def __init__(self, json_data, source_url: str = ""):
+        selected_file = pick_civitai_file_from_version_json(json_data, source_url=source_url)
         self.model_version_id = json_data.get("id", "")
         self.model_id = json_data.get("modelId", "")
-        self.download_url = json_data.get("downloadUrl", "")
+        self.download_url = selected_file.get("downloadUrl", "") or json_data.get("downloadUrl", "")
         self.model_url = f"https://civitai.com/models/{self.model_id}?modelVersionId={self.model_version_id}"
-        self.filename_url = next(
-            (v.get("name", "") for v in reversed(json_data.get("files", [])) if str(self.model_version_id) in v.get("downloadUrl", "")), ""
-        )
-        self.filename_url = self.filename_url if self.filename_url else ""
+        self.filename_url = selected_file.get("name", "") or ""
         self.description = json_data.get("description", "")
-        if self.description is None: self.description = ""
+        if self.description is None:
+            self.description = ""
         self.model_name = json_data.get("model", {}).get("name", "")
         self.model_type = json_data.get("model", {}).get("type", "")
         self.nsfw = json_data.get("model", {}).get("nsfw", False)
@@ -184,71 +573,235 @@ class ModelInformation:
         self.images = [img.get("url", "") for img in json_data.get("images", [])]
         self.example_prompt = json_data.get("trainedWords", [""])[0] if json_data.get("trainedWords") else ""
         self.original_json = copy.deepcopy(json_data)
+        self.selected_file = copy.deepcopy(selected_file)
 
 
 def retrieve_model_info(url):
     json_data = request_json_data(url)
     if not json_data:
         return None
-    model_descriptor = ModelInformation(json_data)
+    model_descriptor = ModelInformation(json_data, source_url=url)
+    filters = get_civitai_query_filters(url)
+    if filters:
+        selected_summary = describe_civitai_file_for_log(model_descriptor.selected_file)
+        if selected_summary:
+            print(f"Civitai selected file: filters={filters} {selected_summary}")
+        else:
+            print(f"Civitai selected file: filters={filters} using model-level downloadUrl")
     return model_descriptor
 
 
+def list_downloaded_candidate_files(directory):
+    try:
+        return {
+            str(path.resolve())
+            for path in Path(directory).iterdir()
+            if path.is_file() and not path.name.endswith(".aria2")
+        }
+    except Exception:
+        return set()
+
+
+def sanitize_civitai_log_text(text: str):
+    output = str(text or "")
+    if not output:
+        return output
+    output = re.sub(r"([?&]token=)[^&\s\"']+", r"\1***", output, flags=re.IGNORECASE)
+    output = re.sub(r"([?&]Authorization=)[^&\s\"']+", r"\1***", output, flags=re.IGNORECASE)
+    return output
+
+
+def build_civitai_aria2_args(directory, download_url: str, filename: str = ""):
+    args = [
+        "aria2c",
+        "--console-log-level=error",
+        "--summary-interval=10",
+        "--user-agent", USER_AGENT,
+        "--referer", CIVITAI_REFERER,
+        "-x", str(CIVITAI_ARIA2_CONNECTIONS),
+        "-k", str(CIVITAI_ARIA2_MIN_SPLIT_SIZE),
+        "-s", str(CIVITAI_ARIA2_SPLIT),
+        "-d", str(directory),
+    ]
+    if CIVITAI_ARIA2_RESUME:
+        args.append("-c")
+    if filename:
+        args.extend(["-o", str(filename)])
+    args.append(str(download_url))
+    return args
+
+
+def run_civitai_aria2(directory, download_url: str, filename: str = ""):
+    args = build_civitai_aria2_args(directory, download_url, filename=filename)
+    result = subprocess.run(args, capture_output=True, text=True)
+    output = "\n".join([part for part in [result.stdout, result.stderr] if part]).strip()
+    return result.returncode, output
+
+
+def classify_civitai_download_failure(output_text: str):
+    text = str(output_text or "")
+    lower = text.lower()
+    if "status=403" in lower and "b2.civitai.com" in lower:
+        return "b2_403"
+    if "status=403" in lower and "civitai.com/api/download/models/" in lower:
+        return "api_403"
+    if "status=403" in lower:
+        return "http_403"
+    if "timed out" in lower or "timeout" in lower:
+        return "timeout"
+    return "other"
+
+
+def cleanup_civitai_download_artifacts(directory, filename: str = ""):
+    removed = []
+    if not filename:
+        return removed
+    target = Path(directory) / filename
+    for candidate in [target, Path(str(target) + ".aria2")]:
+        try:
+            if candidate.exists() and candidate.is_file():
+                candidate.unlink()
+                removed.append(str(candidate))
+        except Exception as e:
+            print(f"[civitai] cleanup failed path={candidate} {type(e).__name__}: {e}")
+    return removed
+
+
+def guess_downloaded_file_path(directory, before_files, expected_filename=""):
+    expected_path = os.path.join(directory, expected_filename) if expected_filename else ""
+    if expected_path and os.path.exists(expected_path):
+        return expected_path
+
+    after_files = list_downloaded_candidate_files(directory)
+    new_files = sorted(list(after_files - set(before_files)))
+    if len(new_files) == 1:
+        return new_files[0]
+
+    if expected_filename:
+        expected_name = str(expected_filename).strip()
+        stem = Path(expected_name).stem
+        suffix = Path(expected_name).suffix.lower()
+        matched = []
+        for path_str in new_files:
+            path_obj = Path(path_str)
+            if suffix and path_obj.suffix.lower() != suffix:
+                continue
+            if stem and (path_obj.stem == stem or path_obj.name == expected_name):
+                matched.append(path_str)
+        if len(matched) == 1:
+            return matched[0]
+
+    return None
+
+
 def download_things(directory, url, hf_token="", civitai_api_key="", romanize=False):
     hf_token = get_token()
     url = url.strip()
     downloaded_file_path = None
 
     if "drive.google.com" in url:
+        before_files = list_downloaded_candidate_files(directory)
         original_dir = os.getcwd()
         os.chdir(directory)
         os.system(f"gdown --fuzzy {url}")
         os.chdir(original_dir)
+        downloaded_file_path = guess_downloaded_file_path(directory, before_files)
     elif "huggingface.co" in url:
         url = url.replace("?download=true", "")
-        # url = urllib.parse.quote(url, safe=':/')  # fix encoding
         if "/blob/" in url:
             url = url.replace("/blob/", "/resolve/")
 
         filename = unidecode(url.split('/')[-1]) if romanize else url.split('/')[-1]
-
         download_hf_file(directory, url, filename, hf_token)
-
         downloaded_file_path = os.path.join(directory, filename)
-
-    elif "civitai.com" in url:
-
+    elif is_civitai_url(url):
         if not civitai_api_key:
-            print("\033[91mYou need an API key to download Civitai models.\033[0m")
-
-        model_profile = retrieve_model_info(url)
-        if model_profile.download_url and model_profile.filename_url:
+            print("You need an API key to download Civitai models.")
+
+        civitai_context = get_civitai_request_context(url, api_key=civitai_api_key)
+        normalized_url = civitai_context["normalized_url"]
+        if normalized_url != url:
+            print(f"Civitai download URL normalized: {sanitize_url_for_log(url)} -> {sanitize_url_for_log(normalized_url)}")
+        model_profile = retrieve_model_info(normalized_url)
+        if model_profile and model_profile.download_url:
             url = model_profile.download_url
-            filename = unidecode(model_profile.filename_url) if romanize else model_profile.filename_url
+            filename = model_profile.filename_url or ""
+            if filename and romanize:
+                filename = unidecode(filename)
         else:
-            if "?" in url:
-                url = url.split("?")[0]
+            url = normalize_civitai_download_api_url(normalized_url)
+            if not is_civitai_download_api_path(get_civitai_url_parts(url).path):
+                print(f"Civitai download URL unresolved: {sanitize_url_for_log(normalized_url)}")
+                return None
             filename = ""
 
-        url_dl = url + f"?token={civitai_api_key}"
-        print(f"Filename: {filename}")
-
-        param_filename = ""
-        if filename:
-            param_filename = f"-o '{filename}'"
-
-        aria2_command = (
-            f'aria2c --console-log-level=error --summary-interval=10 -c -x 16 '
-            f'-k 1M -s 16 -d "{directory}" {param_filename} "{url_dl}"'
-        )
-        os.system(aria2_command)
-
-        if param_filename and os.path.exists(os.path.join(directory, filename)):
-            downloaded_file_path = os.path.join(directory, filename)
+        signed_url = ""
+        try:
+            signed_url = resolve_civitai_download_url(url, civitai_api_key, max_tries=2)
+        except Exception as e:
+            print(f"[civitai] failed to resolve signed download url: {sanitize_url_for_log(url)} {type(e).__name__}: {e}")
+            return None
 
+        signed_host = get_civitai_url_parts(signed_url).netloc
+        print(f"Filename: {filename}")
+        print(f"[civitai] resolved signed host={signed_host or '-'} url={sanitize_url_for_log(url)}")
+
+        before_files = list_downloaded_candidate_files(directory)
+        download_status, download_output = run_civitai_aria2(directory, signed_url, filename=filename)
+        if download_status != 0:
+            failure_kind = classify_civitai_download_failure(download_output)
+            print(
+                f"[civitai] download failed kind={failure_kind} status={download_status} "
+                f"filename={filename or '-'} url={sanitize_url_for_log(url)}"
+            )
+            if download_output:
+                print(sanitize_civitai_log_text(download_output))
+
+            if failure_kind == "b2_403":
+                retry_count = 0
+                while retry_count < CIVITAI_ARIA2_FRESH_RETRY_LIMIT and download_status != 0:
+                    retry_count += 1
+                    removed = cleanup_civitai_download_artifacts(directory, filename=filename)
+                    stale_hint = "yes" if removed or filename else "unknown"
+                    print(
+                        f"[civitai] retrying fresh api/download request after b2_403 "
+                        f"attempt={retry_count}/{CIVITAI_ARIA2_FRESH_RETRY_LIMIT} stale_resume={stale_hint} "
+                        f"filename={filename or '-'}"
+                    )
+                    if removed:
+                        print(f"[civitai] removed stale partials: {removed}")
+                    try:
+                        signed_url = resolve_civitai_download_url(url, civitai_api_key, max_tries=2)
+                        signed_host = get_civitai_url_parts(signed_url).netloc
+                        print(f"[civitai] resolved retry signed host={signed_host or '-'} url={sanitize_url_for_log(url)}")
+                    except Exception as e:
+                        print(f"[civitai] retry resolve failed url={sanitize_url_for_log(url)} error={type(e).__name__}: {e}")
+                        break
+                    download_status, download_output = run_civitai_aria2(directory, signed_url, filename=filename)
+                    if download_status == 0:
+                        print(f"[civitai] download recovered after fresh retry: {filename or sanitize_url_for_log(url)}")
+                        break
+                    retry_kind = classify_civitai_download_failure(download_output)
+                    print(
+                        f"[civitai] retry failed kind={retry_kind} status={download_status} "
+                        f"filename={filename or '-'} url={sanitize_url_for_log(url)}"
+                    )
+                    if download_output:
+                        print(sanitize_civitai_log_text(download_output))
+
+            if download_status != 0:
+                print(f"Civitai download command exited with status {download_status}: {sanitize_url_for_log(url)}")
+
+        downloaded_file_path = guess_downloaded_file_path(directory, before_files, expected_filename=filename)
+        if not downloaded_file_path:
+            print(f"Civitai downloaded file path unresolved: {sanitize_url_for_log(url)}")
     else:
-        os.system(f"aria2c --console-log-level=error --summary-interval=10 -c -x 16 -k 1M -s 16 -d {directory} {url}")
+        before_files = list_downloaded_candidate_files(directory)
+        os.system(f"aria2c --console-log-level=error --summary-interval=10 -c -x 16 -k 1M -s 16 -d \"{directory}\" \"{url}\"")
+        downloaded_file_path = guess_downloaded_file_path(directory, before_files)
 
+    if downloaded_file_path and os.path.exists(downloaded_file_path):
+        print(f"Downloaded file path: {downloaded_file_path}")
     return downloaded_file_path
 
 
@@ -265,13 +818,15 @@ def get_download_file(temp_dir, url, civitai_key="", progress=gr.Progress(track_
     else:
         print(f"Start downloading: {url}")
         before = get_local_model_list(temp_dir)
+        downloaded_path = ""
         try:
-            download_things(temp_dir, url.strip(), HF_TOKEN, civitai_key)
+            downloaded_path = download_things(temp_dir, url.strip(), HF_TOKEN, civitai_key) or ""
         except Exception:
             print(f"Download failed: {url}")
             return ""
         after = get_local_model_list(temp_dir)
-        new_file = list_sub(after, before)[0] if list_sub(after, before) else ""
+        fallback_files = list_sub(after, before)
+        new_file = downloaded_path if downloaded_path and Path(downloaded_path).exists() else (fallback_files[0] if fallback_files else "")
     if not new_file:
         print(f"Download failed: {url}")
         return ""
@@ -576,41 +1131,6 @@ def get_private_lora_model_lists():
 private_lora_model_list = get_private_lora_model_lists()
 
 
-def get_civitai_info(path):
-    global civitai_not_exists_list
-    default = ["", "", "", "", ""]
-    if path in set(civitai_not_exists_list): return default
-    if not Path(path).exists(): return None
-    user_agent = get_user_agent()
-    headers = {'User-Agent': user_agent, 'content-type': 'application/json'}
-    base_url = 'https://civitai.com/api/v1/model-versions/by-hash/'
-    params = {}
-    session = requests.Session()
-    retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
-    session.mount("https://", HTTPAdapter(max_retries=retries))
-    import hashlib
-    with open(path, 'rb') as file:
-        file_data = file.read()
-    hash_sha256 = hashlib.sha256(file_data).hexdigest()
-    url = base_url + hash_sha256
-    try:
-        r = session.get(url, params=params, headers=headers, stream=True, timeout=(3.0, 15))
-    except Exception as e:
-        print(e)
-        return default
-    if not r.ok: return None
-    json = r.json()
-    if not 'baseModel' in json:
-        civitai_not_exists_list.append(path)
-        return default
-    items = []
-    items.append(" / ".join(json['trainedWords']))
-    items.append(json['baseModel'])
-    items.append(json['model']['name'])
-    items.append(f"https://civitai.com/models/{json['modelId']}")
-    items.append(json['images'][0]['url'])
-    return items
-
 
 def get_lora_model_list():
     loras = list_uniq(get_private_lora_model_lists() + DIFFUSERS_FORMAT_LORAS + get_local_model_list(DIRECTORY_LORAS))
@@ -662,28 +1182,58 @@ def update_lora_dict(path):
     loras_dict[key] = items
 
 
+def finalize_downloaded_lora_path(file_path: str, source_url: str = ""):
+    global loras_url_to_path_dict
+    if not file_path:
+        return ""
+    path = Path(file_path)
+    if not path.exists():
+        return ""
+    new_path = Path(f'{path.parent.name}/{escape_lora_basename(path.stem)}{path.suffix}')
+    try:
+        if path.resolve() != new_path.resolve():
+            if new_path.exists():
+                new_path = new_path.resolve()
+            else:
+                new_path = path.resolve().rename(new_path.resolve())
+        else:
+            new_path = path.resolve()
+    except Exception as e:
+        print(f"Failed to normalize downloaded lora path: {file_path} {e}")
+        new_path = path.resolve()
+
+    final_path = str(new_path)
+    if source_url:
+        loras_url_to_path_dict[source_url] = final_path
+        if is_civitai_url(source_url):
+            normalized_url = get_civitai_request_context(source_url, api_key=CIVITAI_API_KEY).get("normalized_url", "")
+            if normalized_url:
+                loras_url_to_path_dict[normalized_url] = final_path
+    update_lora_dict(final_path)
+    return final_path
+
+
 def download_lora(dl_urls: str):
     global loras_url_to_path_dict
     dl_path = ""
-    before = get_local_model_list(DIRECTORY_LORAS)
-    urls = []
-    for url in [url.strip() for url in dl_urls.split(',')]:
-        local_path = f"{DIRECTORY_LORAS}/{url.split('/')[-1]}"
-        if not Path(local_path).exists():
-            download_things(DIRECTORY_LORAS, url, HF_TOKEN, CIVITAI_API_KEY)
-            urls.append(url)
-    after = get_local_model_list(DIRECTORY_LORAS)
-    new_files = list_sub(after, before)
-    i = 0
-    for file in new_files:
-        path = Path(file)
-        if path.exists():
-            new_path = Path(f'{path.parent.name}/{escape_lora_basename(path.stem)}{path.suffix}')
-            path.resolve().rename(new_path.resolve())
-            loras_url_to_path_dict[urls[i]] = str(new_path)
-            update_lora_dict(str(new_path))
-            dl_path = str(new_path)
-        i += 1
+    for url in [url.strip() for url in dl_urls.split(',') if url.strip()]:
+        cached_path = loras_url_to_path_dict.get(url, "")
+        if cached_path and Path(cached_path).exists():
+            dl_path = cached_path
+            continue
+
+        if is_civitai_url(url):
+            normalized_url = get_civitai_request_context(url, api_key=CIVITAI_API_KEY).get("normalized_url", "")
+            cached_path = loras_url_to_path_dict.get(normalized_url, "") if normalized_url else ""
+            if cached_path and Path(cached_path).exists():
+                loras_url_to_path_dict[url] = cached_path
+                dl_path = cached_path
+                continue
+
+        downloaded_path = download_things(DIRECTORY_LORAS, url, HF_TOKEN, CIVITAI_API_KEY)
+        final_path = finalize_downloaded_lora_path(downloaded_path or "", source_url=url)
+        if final_path:
+            dl_path = final_path
     return dl_path
 
 
@@ -746,9 +1296,15 @@ def get_valid_lora_name(query: str, model_name: str):
 
 def get_valid_lora_path(query: str):
     path = None
-    if not query or query == "None": return None
-    if to_lora_key(query) in loras_dict.keys(): return query
-    if Path(path).exists():
+    if not query or query == "None":
+        return None
+    if to_lora_key(query) in loras_dict.keys():
+        return query
+    if query in loras_url_to_path_dict.keys():
+        path = loras_url_to_path_dict[query]
+    else:
+        path = to_lora_path(query.strip().split('/')[-1])
+    if path and Path(path).exists():
         return path
     else:
         return None
@@ -797,7 +1353,7 @@ def set_prompt_loras(prompt, prompt_syntax, model_name, lora1, lora1_wt, lora2,
             wt = result[0][1]
             path = to_lora_path(key)
             if not key in loras_dict.keys() or not Path(path).exists():
-                path = get_valid_lora_name(path)
+                path = get_valid_lora_name(path, model_name)
                 if not path or path == "None": continue
             if path in lora_paths or key in lora_paths:
                 continue
@@ -1037,84 +1593,109 @@ CIVITAI_FILETYPE = ["Model", "VAE", "Config", "Training Data"]
 def get_civitai_info(path):
     global civitai_not_exists_list, loras_url_to_path_dict
     default = ["", "", "", "", ""]
-    if path in set(civitai_not_exists_list): return default
-    if not Path(path).exists(): return None
-    user_agent = get_user_agent()
-    headers = {'User-Agent': user_agent, 'content-type': 'application/json'}
+    if path in set(civitai_not_exists_list):
+        return default
+    if not Path(path).exists():
+        return None
+
+    headers = get_civitai_headers(CIVITAI_API_KEY)
     base_url = 'https://civitai.com/api/v1/model-versions/by-hash/'
-    params = {}
-    session = requests.Session()
-    retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
-    session.mount("https://", HTTPAdapter(max_retries=retries))
+    session = create_retry_session()
+
     import hashlib
+    sha256_hash = hashlib.sha256()
     with open(path, 'rb') as file:
-        file_data = file.read()
-    hash_sha256 = hashlib.sha256(file_data).hexdigest()
+        for chunk in iter(lambda: file.read(1024 * 1024), b''):
+            sha256_hash.update(chunk)
+    hash_sha256 = sha256_hash.hexdigest()
     url = base_url + hash_sha256
     try:
-        r = session.get(url, params=params, headers=headers, stream=True, timeout=(3.0, 15))
+        r = session.get(url, headers=headers, stream=True, timeout=CIVITAI_METADATA_TIMEOUT)
     except Exception as e:
-        print(e)
+        print(f"Civitai by-hash lookup failed: {path} {type(e).__name__}: {e}")
         return default
-    else:
-        if not r.ok: return None
-        json = r.json()
-        if 'baseModel' not in json:
+    if not r.ok:
+        print(f"Civitai by-hash lookup status={r.status_code}: {path}")
+        if r.status_code == 404:
             civitai_not_exists_list.append(path)
             return default
-        items = []
-        items.append(" / ".join(json['trainedWords']))                  # The words (prompts) used to trigger the model
-        items.append(json['baseModel'])                                 # Base model (SDXL1.0, Pony, ...)
-        items.append(json['model']['name'])                             # The name of the model version
-        items.append(f"https://civitai.com/models/{json['modelId']}")   # The repo url for the model
-        items.append(json['images'][0]['url'])                          # The url for a sample image
-        loras_url_to_path_dict[path] = json['downloadUrl']              # The download url to get the model file for this specific version
-        return items
+        return None
+    try:
+        json_data = r.json()
+    except Exception as e:
+        print(f"Civitai by-hash JSON parse failed: {path} {type(e).__name__}: {e}")
+        return default
+    if 'baseModel' not in json_data:
+        civitai_not_exists_list.append(path)
+        return default
+
+    selected_file = pick_civitai_file_from_version_json(json_data, source_url=json_data.get('downloadUrl', ''))
+    items = []
+    items.append(" / ".join(json_data.get('trainedWords', [])))
+    items.append(json_data.get('baseModel', ''))
+    items.append(json_data.get('model', {}).get('name', ''))
+    items.append(f"https://civitai.com/models/{json_data.get('modelId', '')}")
+    images = json_data.get('images', []) if isinstance(json_data.get('images'), list) else []
+    items.append(images[0].get('url', '') if images else '')
+    download_url = selected_file.get('downloadUrl', '') or json_data.get('downloadUrl', '')
+    if download_url:
+        loras_url_to_path_dict[path] = normalize_civitai_download_api_url(download_url)
+    return items
 
 
 def search_lora_on_civitai(query: str, allow_model: list[str] = ["Pony", "SDXL 1.0"], limit: int = 100,
                            sort: str = "Highest Rated", period: str = "AllTime", tag: str = "", user: str = "", page: int = 1):
-    user_agent = get_user_agent()
-    headers = {'User-Agent': user_agent, 'content-type': 'application/json'}
-    if CIVITAI_API_KEY: headers['Authorization'] = f'Bearer {{{CIVITAI_API_KEY}}}'
+    headers = get_civitai_headers(CIVITAI_API_KEY)
     base_url = 'https://civitai.com/api/v1/models'
     params = {'types': ['LORA'], 'sort': sort, 'period': period, 'limit': limit, 'page': int(page), 'nsfw': 'true'}
-    if query: params["query"] = query
-    if tag: params["tag"] = tag
-    if user: params["username"] = user
-    session = requests.Session()
-    retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
-    session.mount("https://", HTTPAdapter(max_retries=retries))
+    if query:
+        params["query"] = query
+    if tag:
+        params["tag"] = tag
+    if user:
+        params["username"] = user
+    session = create_retry_session()
     try:
-        r = session.get(base_url, params=params, headers=headers, stream=True, timeout=(3.0, 30))
+        r = session.get(base_url, params=params, headers=headers, stream=True, timeout=CIVITAI_SEARCH_TIMEOUT)
     except Exception as e:
-        print(e)
+        print(f"Civitai search failed: query={query!r} page={page} {type(e).__name__}: {e}")
         return None
-    else:
-        if not r.ok: return None
-        json = r.json()
-        if 'items' not in json: return None
-        items = []
-        for j in json['items']:
-            for model in j['modelVersions']:
-                item = {}
-                if len(allow_model) != 0 and model['baseModel'] not in set(allow_model): continue
-                item['name'] = j['name']
-                item['creator'] = j['creator']['username'] if 'creator' in j.keys() and 'username' in j['creator'].keys() else ""
-                item['tags'] = j['tags'] if 'tags' in j.keys() else []
-                item['model_name'] = model['name'] if 'name' in model.keys() else ""
-                item['base_model'] = model['baseModel'] if 'baseModel' in model.keys() else ""
-                item['description'] = model['description'] if 'description' in model.keys() else ""
-                item['dl_url'] = model['downloadUrl']
-                item['md'] = ""
-                if 'images' in model.keys() and len(model["images"]) != 0:
-                    item['img_url'] = model["images"][0]["url"]
-                    item['md'] += f'<img src="{model["images"][0]["url"]}#float" alt="thumbnail" width="150" height="240"><br>'
-                else: item['img_url'] = "/home/user/app/null.png"
-                item['md'] += f'''Model URL: [https://civitai.com/models/{j["id"]}](https://civitai.com/models/{j["id"]})<br>Model Name: {item["name"]}<br>
-                    Creator: {item["creator"]}<br>Tags: {", ".join(item["tags"])}<br>Base Model: {item["base_model"]}<br>Description: {item["description"]}'''
-                items.append(item)
-        return items
+    if not r.ok:
+        print(f"Civitai search status={r.status_code}: query={query!r} page={page}")
+        return None
+    json = r.json()
+    if 'items' not in json:
+        print(f"Civitai search returned no items key: query={query!r} page={page}")
+        return None
+    items = []
+    for j in json['items']:
+        model_versions = j.get('modelVersions') if isinstance(j, dict) else []
+        if not isinstance(model_versions, list):
+            continue
+        for model in model_versions:
+            if not isinstance(model, dict):
+                continue
+            base_model = model.get('baseModel', '')
+            if len(allow_model) != 0 and base_model not in set(allow_model):
+                continue
+            item = {}
+            item['name'] = j['name'] if isinstance(j, dict) and 'name' in j else ""
+            item['creator'] = j['creator']['username'] if isinstance(j, dict) and 'creator' in j and isinstance(j['creator'], dict) and 'username' in j['creator'] else ""
+            item['tags'] = j['tags'] if isinstance(j, dict) and 'tags' in j else []
+            item['model_name'] = model.get('name', '')
+            item['base_model'] = base_model
+            item['description'] = model.get('description', '')
+            item['dl_url'] = model.get('downloadUrl', '')
+            item['md'] = ""
+            if 'images' in model.keys() and len(model["images"]) != 0:
+                item['img_url'] = model["images"][0]["url"]
+                item['md'] += f'<img src="{model["images"][0]["url"]}#float" alt="thumbnail" width="150" height="240"><br>'
+            else:
+                item['img_url'] = "/home/user/app/null.png"
+            item['md'] += f'''Model URL: [https://civitai.com/models/{j["id"]}](https://civitai.com/models/{j["id"]})<br>Model Name: {item["name"]}<br>
+                Creator: {item["creator"]}<br>Tags: {", ".join(item["tags"])}<br>Base Model: {item["base_model"]}<br>Description: {item["description"]}'''
+            items.append(item)
+    return items
 
 
 def search_civitai_lora(query, base_model=[], sort=CIVITAI_SORT[0], period=CIVITAI_PERIOD[0], tag="", user="", gallery=[]):