"""Hugging Face inspection + downloads (with progress), model manifests, verification.""" from __future__ import annotations import asyncio import json import logging import os import re import shutil import time from pathlib import Path from typing import Any from .config import Settings from .errors import APIError, DownloadError, InsufficientDisk from .jobs import Job, JobRunner from .models import compat, formats from .models.estimator import kv_bytes_per_token from .models.scanner import slugify log = logging.getLogger("llm_api.downloads") GB = 1024**3 REPO_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,95}/[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") TRUSTED_GGUF_AUTHORS = {"unsloth", "bartowski", "ggml-org", "lmstudio-community", "Qwen", "google", "mistralai", "microsoft", "TheBloke", "mradermacher", "QuantFactory", "nomic-ai", "BAAI", "jinaai", "mixedbread-ai", "second-state", "openai", "deepseek-ai", "zai-org", "nvidia", "meta-llama"} def parse_repo(text: str) -> str: t = text.strip() m = re.match(r"^https?://huggingface\.co/([^/\s]+/[^/\s?#]+)", t) if m: t = m.group(1) t = t.removeprefix("hf.co/").removeprefix("huggingface.co/") if not REPO_RE.match(t) or ".." in t: raise APIError("Invalid Hugging Face repository id. Expected 'organization/model-name'.", param="repository") return t def _quant_rank(label: str | None) -> float: if not label: return 0 _, bits = formats.parse_quant_from_name("x-" + label) return bits or 0 def pick_gguf_files(files: list[dict], preferred: str | None = None) -> list[dict]: """Choose one quantization from a GGUF repo (prefer Q4_K_M/Q5_K_M/Q6_K/Q8_0, or preferred).""" ggufs = [f for f in files if f["path"].lower().endswith(".gguf")] if not ggufs: return [] mmproj = [f for f in ggufs if "mmproj" in f["path"].lower()] weights = [f for f in ggufs if "mmproj" not in f["path"].lower()] def label(f): q, _ = formats.parse_quant_from_name(Path(f["path"]).stem) return (q or "").upper() groups: dict[str, list[dict]] = {} for f in weights: base = re.sub(r"-\d{5}-of-\d{5}$", "", Path(f["path"]).stem) groups.setdefault(base, []).append(f) if preferred: pref = preferred.upper() for base, fs in groups.items(): if pref in base.upper(): return sorted(fs, key=lambda f: f["path"]) + mmproj[:1] order = ["Q4_K_M", "Q4_K_XL", "UD-Q4_K_XL", "Q5_K_M", "Q4_K_S", "Q6_K", "Q8_0", "MXFP4", "Q5_K_S", "IQ4_XS", "Q4_0", "BF16", "F16"] for want in order: for base, fs in groups.items(): if want in base.upper(): return sorted(fs, key=lambda f: f["path"]) + mmproj[:1] if not groups: return [] base = sorted(groups.items(), key=lambda kv: sum(f["size"] or 0 for f in kv[1]))[0] return sorted(base[1], key=lambda f: f["path"]) + mmproj[:1] class Downloader: def __init__(self, settings: Settings, db, registry, jobs: JobRunner): self.settings = settings self.db = db self.registry = registry self.jobs = jobs def _api(self): from huggingface_hub import HfApi return HfApi(token=self.settings.hf_token or os.environ.get("HF_TOKEN") or None) # ------------------------------------------------------------------ inspect async def inspect(self, repo: str, quant: str | None = None) -> dict: repo = parse_repo(repo) api = self._api() try: info = await asyncio.to_thread(api.model_info, repo, files_metadata=True) except Exception as e: msg = str(e) if "401" in msg or "gated" in msg.lower(): raise DownloadError(f"Repository '{repo}' is gated or private. Set HF_TOKEN and accept the license on Hugging Face.") if "404" in msg: raise DownloadError(f"Repository '{repo}' was not found on Hugging Face.") if "429" in msg: raise DownloadError("Hugging Face rate limit reached. Try again in a minute.") raise DownloadError(f"Could not inspect '{repo}': {msg[:200]}") files = [{"path": s.rfilename, "size": s.size or 0} for s in (info.siblings or [])] tags = list(info.tags or []) library = getattr(info, "library_name", None) cfg = getattr(info, "config", None) or {} has_st = any(f["path"].endswith(".safetensors") for f in files) has_gguf = any(f["path"].lower().endswith(".gguf") for f in files) is_mlx = "mlx" in tags or library == "mlx" runtime = None selected: list[dict] = [] if has_gguf: runtime = "llamacpp" selected = pick_gguf_files(files, quant) weights_bytes = sum(f["size"] for f in selected if "mmproj" not in f["path"].lower()) download_bytes = sum(f["size"] for f in selected) elif has_st and is_mlx: runtime = "mlx" selected = [f for f in files if not f["path"].endswith((".gguf", ".bin", ".pt", ".onnx", ".h5", ".msgpack")) and ".cache" not in f["path"]] weights_bytes = sum(f["size"] for f in selected if f["path"].endswith(".safetensors")) download_bytes = sum(f["size"] for f in selected) elif has_st: runtime = "mlx" # mlx_lm can load standard HF safetensors (quantizes nothing; runs bf16) selected = [f for f in files if not f["path"].endswith((".gguf", ".bin", ".pt", ".onnx", ".h5", ".msgpack"))] weights_bytes = sum(f["size"] for f in selected if f["path"].endswith(".safetensors")) download_bytes = sum(f["size"] for f in selected) else: raise DownloadError(f"Repository '{repo}' contains neither safetensors nor GGUF weights.") parsed = formats.parse_hf_config(cfg) if cfg else {} model_type = parsed.get("model_type") base_model = next((t.split(":", 1)[1] for t in tags if t.startswith("base_model:") and "quantized:" not in t and "finetune:" not in t), None) bm_quant = next((t.split(":", 2)[2] for t in tags if t.startswith("base_model:quantized:")), None) quant_label = parsed.get("quantization") bits = parsed.get("quant_bits") if runtime == "llamacpp" and selected: quant_label, bits = formats.parse_quant_from_name(Path(selected[0]["path"]).stem) if not quant_label: quant_label, bits = formats.parse_quant_from_name(repo.split("/")[-1]) if not quant_label and runtime == "mlx": quant_label, bits = (parsed.get("torch_dtype") or "bf16"), 16 total_p, active_p = formats.parse_param_count_from_name(repo.split("/")[-1]) if not total_p and weights_bytes and bits: total_p = int(weights_bytes * 8 / bits) kv = kv_bytes_per_token(parsed.get("n_layers"), parsed.get("n_kv_heads"), parsed.get("head_dim"), 16, parsed.get("full_attention_layers"), parsed.get("sliding_window")) if not kv and total_p: # crude fallback: ~ 130 KB/token for 7-9B, scale with params^0.6 kv = int(130_000 * (total_p / 8e9) ** 0.6) pipeline = getattr(info, "pipeline_tag", None) name_l = repo.lower() vision = bool(parsed.get("vision")) or pipeline == "image-text-to-text" or any("mmproj" in f["path"].lower() for f in selected) embedding = pipeline in ("feature-extraction", "sentence-similarity") or "embed" in name_l reranker = pipeline == "text-ranking" or "rerank" in name_l budget, absolute = await self.registry.budgets() from .models.scanner import llamacpp_available comp = compat.evaluate(runtime=runtime, weights_bytes=weights_bytes, kv_per_token=kv, max_context=parsed.get("max_context"), model_type=model_type, architecture=(parsed.get("architectures") or [None])[0] if runtime == "mlx" else model_type, vision=vision, embedding=embedding, reranker=reranker, budget_gb=budget, absolute_gb=absolute, llamacpp_available=llamacpp_available(self.settings.llama_server_bin), quant_bits=bits, weights_file=selected[0]["path"] if runtime == "llamacpp" and selected else None) if runtime == "llamacpp" and comp.status == compat.INCOMPATIBLE and "Architecture" in comp.reason: pass du = shutil.disk_usage(self.settings.models_dir) min_free = float(await self.db.get_setting("min_free_disk_gb", self.settings.min_free_disk_gb)) free_after = (du.free - download_bytes) / GB target = self.target_dir(repo, runtime, vision, embedding, reranker) existing = await self.db.fetchone("SELECT id FROM models WHERE repository=? AND installed=1", (repo,)) return { "repository": repo, "runtime": runtime, "format": "gguf" if runtime == "llamacpp" else "safetensors", "files": selected, "all_files": files, "download_bytes": download_bytes, "weights_bytes": weights_bytes, "quantization": quant_label, "quant_bits": bits, "parameter_count": total_p, "active_parameter_count": active_p, "model_type": model_type, "pipeline_tag": pipeline, "library": library, "tags": tags[:40], "base_model": base_model or bm_quant, "vision": vision, "embedding": embedding, "reranker": reranker, "max_context": parsed.get("max_context"), "kv_bytes_per_token": kv, "compatibility": comp.to_dict(), "size_class": formats.size_class(comp.estimated_ram_gb, budget), "disk": {"free_gb": round(du.free / GB, 1), "free_after_gb": round(free_after, 1), "min_free_gb": min_free, "ok": free_after >= min_free}, "target_dir": str(target), "already_installed": existing["id"] if existing else None, "downloads": getattr(info, "downloads", None), "likes": getattr(info, "likes", None), "last_modified": str(getattr(info, "last_modified", "") or ""), "gated": bool(getattr(info, "gated", False)), } def target_dir(self, repo: str, runtime: str, vision: bool, embedding: bool, reranker: bool) -> Path: name = repo.split("/")[-1] root = self.settings.models_dir if embedding: sub = root / "embeddings" elif reranker: sub = root / "rerankers" elif vision: sub = root / "vision" else: sub = root / ("gguf" if runtime == "llamacpp" else "mlx") family = formats.guess_family(name) return sub / family / name # ------------------------------------------------------------------ download async def start_download(self, repo: str, quant: str | None = None, *, force: bool = False, actor: str | None = None) -> Job: if not await self.db.get_setting("allow_downloads", self.settings.allow_downloads): raise APIError("Downloads are disabled in settings.", code="DOWNLOADS_DISABLED", status_code=403) insp = await self.inspect(repo, quant) if insp["already_installed"] and not force: raise APIError(f"'{repo}' is already installed as '{insp['already_installed']}'.", code="ALREADY_INSTALLED", status_code=409) if not insp["disk"]["ok"]: raise InsufficientDisk(f"Downloading {insp['download_bytes'] / GB:.1f} GB would leave {insp['disk']['free_after_gb']} GB free, " f"below the {insp['disk']['min_free_gb']:.0f} GB reserve.") if insp["compatibility"]["status"] == compat.INCOMPATIBLE and not force: raise APIError(f"'{repo}' is incompatible: {insp['compatibility']['reason']}", code="MODEL_INCOMPATIBLE", status_code=422) target = Path(insp["target_dir"]) title = f"Download {repo}" payload = {"repository": repo, "quant": quant, "target": str(target), "runtime": insp["runtime"], "download_bytes": insp["download_bytes"], "files": [f["path"] for f in insp["files"]]} await self.db.audit("download.start", actor=actor, target=repo, detail={"bytes": insp["download_bytes"]}) async def run(job: Job): return await self._download_job(job, insp, target) return self.jobs.submit("download", title, payload, run, exclusive_download=True) async def _download_job(self, job: Job, insp: dict, target: Path) -> dict: from huggingface_hub import hf_hub_download repo = insp["repository"] files = insp["files"] total = max(1, insp["download_bytes"]) target.mkdir(parents=True, exist_ok=True) token = self.settings.hf_token or os.environ.get("HF_TOKEN") or None t0 = time.time() done_bytes = 0 # progress poller: sums sizes of files (+ partial .incomplete blobs) in target stop = asyncio.Event() def measure() -> int: n = 0 for p in target.rglob("*"): if p.is_file(): try: n += p.stat().st_size except OSError: pass return n async def poll(): last = 0 last_t = time.time() while not stop.is_set(): cur = await asyncio.to_thread(measure) now = time.time() speed = (cur - last) / max(0.001, now - last_t) last, last_t = cur, now eta = (total - cur) / speed if speed > 0 else None self.jobs.update(job, progress=min(0.99, cur / total), downloaded=cur, total=total, speed_bps=round(speed), eta_seconds=round(eta) if eta else None, elapsed=round(now - t0)) await asyncio.sleep(1.0) poller = asyncio.create_task(poll()) try: for i, f in enumerate(files): if job.cancelled: break self.jobs.update(job, current_file=f["path"], file_index=i + 1, file_count=len(files)) await asyncio.to_thread(hf_hub_download, repo, f["path"], local_dir=str(target), token=token, force_download=False) done_bytes += f["size"] finally: stop.set() poller.cancel() if job.cancelled: # remove partial download shutil.rmtree(target, ignore_errors=True) return {"cancelled": True} # cleanup HF metadata cache folder inside local_dir shutil.rmtree(target / ".cache", ignore_errors=True) # verify missing = [f["path"] for f in files if not (target / f["path"]).exists()] bad = [f["path"] for f in files if (target / f["path"]).exists() and f["size"] and (target / f["path"]).stat().st_size != f["size"]] if missing or bad: raise DownloadError(f"Download incomplete: missing {missing[:3]} size-mismatch {bad[:3]}") manifest = { "schema_version": 1, "model_id": slugify(repo.split("/")[-1]), "runtime": insp["runtime"], "quantization": insp["quantization"], "download_source": "huggingface", "repository": repo, "downloaded_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "verified": True, "files": [{"path": f["path"], "size": f["size"]} for f in files], "provider": repo.split("/")[0], "base_model": insp.get("base_model"), "task": "embedding" if insp["embedding"] else "reranking" if insp["reranker"] else "image-text-to-text" if insp["vision"] else "text-generation", } (target / "llm-api.json").write_text(json.dumps(manifest, indent=2)) # keep a copy in models/manifests mdir = self.settings.models_dir / "manifests" mdir.mkdir(parents=True, exist_ok=True) (mdir / f"{slugify(repo)}.json").write_text(json.dumps(manifest, indent=2)) self.jobs.update(job, progress=0.99, stage="registering") summary = await self.registry.rescan() m = await self.db.fetchone("SELECT id FROM models WHERE path=?", (str(target),)) mid = m["id"] if m else None if mid: await self.registry.update(mid, repository=repo, provider=repo.split("/")[0], verified=False) if insp.get("embedding") or insp.get("reranker") or insp.get("vision"): ov = {"task": manifest["task"], "vision": insp["vision"], "embedding": insp["embedding"], "reranker": insp["reranker"]} await self.registry.update(mid, overrides=ov, task=manifest["task"], vision=insp["vision"], embedding=insp["embedding"], reranker=insp["reranker"]) await self.db.model_event(mid, "downloaded", {"repository": repo, "bytes": insp["download_bytes"], "seconds": round(time.time() - t0)}) return {"model_id": mid, "path": str(target), "bytes": insp["download_bytes"], "seconds": round(time.time() - t0), "scan": summary} # ------------------------------------------------------------------ delete async def delete_model(self, model_id: str, *, actor: str | None = None, keep_benchmarks: bool = True) -> dict: m = await self.registry.get(model_id) if not m: from .errors import ModelNotFound raise ModelNotFound(f"Model '{model_id}' not found.") path = Path(m["path"]).resolve() root = self.settings.models_dir.resolve() if root not in path.parents: raise APIError("Refusing to delete a path outside the model root.", code="PATH_NOT_ALLOWED", status_code=403) size = m["disk_size_bytes"] if m["format"] == "gguf" and m["weights_file"]: # delete only this quantization's files (+ mmproj if no other model uses the dir) others = await self.db.fetchall("SELECT id FROM models WHERE path=? AND id<>?", (m["path"], model_id)) wf = Path(m["weights_file"]) base = re.sub(r"-\d{5}-of-\d{5}$", "", wf.stem) for f in path.glob("*.gguf"): if f.stem == wf.stem or f.stem.startswith(base + "-0"): f.unlink(missing_ok=True) if not others: shutil.rmtree(path, ignore_errors=True) else: shutil.rmtree(path, ignore_errors=True) await self.registry.delete(model_id) if not keep_benchmarks: await self.db.execute("DELETE FROM model_benchmarks WHERE model_id=?", (model_id,)) await self.db.audit("model.delete", actor=actor, target=model_id, detail={"bytes": size, "path": str(path)}) await self.db.model_event(model_id, "deleted", {"bytes": size}) return {"deleted": model_id, "bytes": size}