Enhance CLI and documentation for capturing VM inventory

- Introduced new `capture` commands in the CLI to allow users to merge VM inventory into local manifests without downloading weights.
- Updated `README.md` and `cli.md` to include detailed instructions for the new capture functionality, including options for models and extensions.
- Enhanced `decisions.md` to clarify the role of captured links in the manifest files.
- Improved `extensions.md` to document the process of capturing installed extensions back to the local configuration.
- Added new functions in `civitai.py` to support fetching model versions by hash and generating canonical URLs for models.
This commit is contained in:
Leonid Pershin
2026-08-21 05:56:52 +03:00
parent 71f4e4c2e3
commit 603165a4ba
10 changed files with 940 additions and 3 deletions
+66 -1
View File
@@ -92,7 +92,8 @@ def fetch_model_version(token: str, host: str, version_id: int, timeout: float =
url = f"https://{candidate}/api/v1/model-versions/{version_id}"
try:
with httpx.Client(timeout=timeout, follow_redirects=True) as client:
response = client.get(url, headers={"Authorization": f"Bearer {token}"})
headers = {"Authorization": f"Bearer {token}"} if token else {}
response = client.get(url, headers=headers)
except httpx.HTTPError as exc:
last_error = str(exc)
continue
@@ -106,3 +107,67 @@ def fetch_model_version(token: str, host: str, version_id: int, timeout: float =
if response.status_code not in {404, 400}:
break
raise CloudError(f"Civitai version {version_id}: {last_error} (хосты {', '.join(seen)})")
def fetch_model_version_by_hash(
token: str | None,
host: str,
sha256: str,
timeout: float = 30.0,
) -> tuple[str, dict]:
"""GET /api/v1/model-versions/by-hash/{sha}; public, token optional for NSFW/region."""
digest = sha256.strip().lower()
if len(digest) != 64 or any(c not in "0123456789abcdef" for c in digest):
raise CloudError(f"Civitai by-hash: неверный SHA256 ({sha256[:16]}…)")
first = _normalize_host(host)
order = [first, other_host(first)]
last_error = "нет ответа"
seen: set[str] = set()
headers = {"Authorization": f"Bearer {token}"} if token else {}
for candidate in order:
if candidate in seen or candidate not in ALLOWED_HOSTS:
continue
seen.add(candidate)
url = f"https://{candidate}/api/v1/model-versions/by-hash/{digest}"
try:
with httpx.Client(timeout=timeout, follow_redirects=True) as client:
response = client.get(url, headers=headers)
except httpx.HTTPError as exc:
last_error = str(exc)
continue
if response.status_code == 200:
data = response.json()
if isinstance(data, dict) and data.get("id") is not None:
return candidate, data
last_error = "пустой ответ"
continue
last_error = f"HTTP {response.status_code}"
if response.status_code not in {404, 400}:
break
raise CloudError(f"Civitai by-hash {digest[:12]}…: {last_error} (хосты {', '.join(seen)})")
def civitai_model_url(model_id: int, version_id: int, host: str = "civitai.red") -> str:
"""Canonical manifest URL (links only — no download)."""
h = _normalize_host(host)
if h not in ALLOWED_HOSTS:
h = "civitai.red"
if h == "civitai.green":
h = "civitai.com"
return f"https://{h}/models/{int(model_id)}?modelVersionId={int(version_id)}"
def version_ids_from_payload(version: dict) -> tuple[int | None, int | None]:
"""Extract (version_id, model_id) from a Civitai version JSON object."""
try:
vid = int(version["id"]) if version.get("id") is not None else None
except (TypeError, ValueError, KeyError):
vid = None
mid = version.get("modelId")
if mid is None and isinstance(version.get("model"), dict):
mid = version["model"].get("id")
try:
model_id = int(mid) if mid is not None else None
except (TypeError, ValueError):
model_id = None
return vid, model_id