first commit

src/gpu_rent/manifests.py

@@ -0,0 +1,102 @@
+"""Parse models.yaml / extensions.yaml. version_id 0 is a placeholder."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from gpu_rent.errors import ConfigError
+
+MODEL_TYPES = (
+    "checkpoint",
+    "lora",
+    "vae",
+    "embedding",
+    "controlnet",
+    "upscaler",
+)
+
+
+@dataclass
+class ModelEntry:
+    kind: str
+    version_id: int | None
+    url: str | None
+
+
+@dataclass
+class GitRepo:
+    kind: str
+    url: str
+    ref: str
+    directory: str | None
+
+
+def _load_yaml(path: Path) -> Any:
+    if not path.is_file():
+        return None
+    text = path.read_text(encoding="utf-8")
+    if not text.strip():
+        return {}
+    try:
+        return yaml.safe_load(text)
+    except yaml.YAMLError as exc:
+        raise ConfigError(f"Не разобрать YAML {path}: {exc}") from exc
+
+
+def parse_models(path: Path) -> list[ModelEntry]:
+    data = _load_yaml(path)
+    if data is None:
+        return []
+    if data == {} or data is None:
+        return []
+    if not isinstance(data, dict):
+        raise ConfigError(f"{path}: корень должен быть mapping типов моделей")
+    entries: list[ModelEntry] = []
+    for kind, items in data.items():
+        if kind not in MODEL_TYPES:
+            continue
+        if not items:
+            continue
+        if not isinstance(items, list):
+            raise ConfigError(f"{path}: {kind} должен быть списком")
+        for item in items:
+            if not isinstance(item, dict):
+                raise ConfigError(f"{path}: элемент {kind} — объект с version_id или url")
+            vid = item.get("version_id")
+            url = item.get("url")
+            if vid in (0, "0", None) and not url:
+                continue
+            version_id = int(vid) if vid not in (None, "", 0, "0") else None
+            entries.append(ModelEntry(kind=kind, version_id=version_id, url=str(url) if url else None))
+    return entries
+
+
+def parse_extensions(path: Path) -> list[GitRepo]:
+    data = _load_yaml(path)
+    if not data:
+        return []
+    if not isinstance(data, dict):
+        raise ConfigError(f"{path}: корень swarmui: / comfy:")
+    repos: list[GitRepo] = []
+    for kind in ("swarmui", "comfy"):
+        items = data.get(kind) or []
+        if not items:
+            continue
+        if not isinstance(items, list):
+            raise ConfigError(f"{path}: {kind} должен быть списком")
+        for item in items:
+            if not isinstance(item, dict) or not item.get("url"):
+                raise ConfigError(f"{path}: у {kind} нужен url")
+            repos.append(
+                GitRepo(
+                    kind=kind,
+                    url=str(item["url"]),
+                    ref=str(item.get("ref") or "main"),
+                    directory=str(item["dir"]) if item.get("dir") else None,
+                )
+            )
+    return repos