"""Where the programs and the models come from: GitHub releases and Hugging Face. Both answer plain JSON over HTTPS without a key, and both publish a sha256 for every file they hand out: GitHub as the asset digest, Hugging Face as the LFS object id. Nothing that lands on disk is trusted for having arrived, which matters more here than it usually would, because half of what is fetched is a program Dikte then runs. The lists are read rather than kept. A model catalogue written into the source means a release of Dikte for every new model, and a pinned whisper.cpp version means one for every whisper.cpp release; both of those are somebody else's news, not Dikte's. Answers are cached for a few hours, and a cache that has gone stale is still a better answer than none when the network is down. Nothing here imports the rest of Dikte apart from the string table: this module knows two websites and nothing about dictation. """ import collections import json import os import pathlib import time import urllib.error import urllib.parse import urllib.request from i18n import t GITHUB_API = "https://api.github.com" HF_API = "https://huggingface.co/api" HF_FILES = "https://huggingface.co" USER_AGENT = "dikte/1.0 (+https://github.com/yusufipk/dikte)" CACHE_DIR = (pathlib.Path(os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache")) / "dikte") # Long enough that opening the settings window twice in an evening asks nobody # anything, short enough that a model published this morning is offered today. CACHE_TTL = 6 * 3600 # `sha256` is empty for the few files neither side stores in LFS; those are the # small ones, and a checksum is only worth having where there is something to # check. Item = collections.namedtuple("Item", "name url size sha256") Repo = collections.namedtuple("Repo", "id downloads updated") class HubError(Exception): pass def _get(url, timeout=20): request = urllib.request.Request(url, headers={ "User-Agent": USER_AGENT, "Accept": "application/json", }) try: with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: exc.close() # it holds the response body open until it is collected raise HubError(t("{url} answered HTTP {code}.", url=urllib.parse.urlsplit(url).netloc, code=exc.code)) from exc except urllib.error.URLError as exc: raise HubError(t("Could not reach {url}: {error}", url=urllib.parse.urlsplit(url).netloc, error=exc.reason)) from exc except (ValueError, OSError) as exc: raise HubError(t("Could not read the answer from {url}: {error}", url=urllib.parse.urlsplit(url).netloc, error=exc)) from exc def _cache_file(key): safe = "".join(c if c.isalnum() or c in "-._" else "-" for c in key) return CACHE_DIR / f"{safe}.json" def _read_cache(key, ttl): """What was stored under this key, or None. `ttl` of 0 ignores the age.""" path = _cache_file(key) try: age = time.time() - path.stat().st_mtime if ttl and age > ttl: return None return json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): return None def _write_cache(key, payload): try: CACHE_DIR.mkdir(parents=True, exist_ok=True) _cache_file(key).write_text(json.dumps(payload), encoding="utf-8") except OSError: pass # a cache that cannot be written is not a failed lookup def _fetch(key, url, ttl=CACHE_TTL, refresh=False): """The JSON at `url`, from the cache when it is fresh enough. A lookup that fails falls back to the cache however old it is: an offline settings window that shows yesterday's list is worth a great deal more than one that shows an error. """ if not refresh: cached = _read_cache(key, ttl) if cached is not None: return cached try: payload = _get(url) except HubError: stale = _read_cache(key, 0) if stale is not None: return stale raise _write_cache(key, payload) return payload def _digest(value): """GitHub writes its digests as "sha256:…"; Hugging Face writes the hash.""" value = (value or "").strip() return value.split(":", 1)[1] if value.startswith("sha256:") else value def release(repo, tag="latest", refresh=False): """(tag, [Item]) for one GitHub release, newest when no tag is given.""" where = "latest" if tag in ("", "latest") else f"tags/{tag}" data = _fetch(f"gh-{repo}-{tag or 'latest'}", f"{GITHUB_API}/repos/{repo}/releases/{where}", refresh=refresh) if not isinstance(data, dict) or not data.get("assets"): raise HubError(t("{repo} has no downloadable release.", repo=repo)) assets = [Item(a.get("name") or "", a.get("browser_download_url") or "", int(a.get("size") or 0), _digest(a.get("digest"))) for a in data["assets"] if a.get("browser_download_url")] return data.get("tag_name") or tag, assets def files(repo, revision="main", refresh=False): """[Item] for every file in a Hugging Face repository. The size is there whether or not the file is in LFS; the hash is only there when it is, which for anything worth downloading it always is. """ data = _fetch(f"hf-tree-{repo}-{revision}", f"{HF_API}/models/{repo}/tree/{revision}?recursive=true", refresh=refresh) if not isinstance(data, list): raise HubError(t("{repo} did not return a file list.", repo=repo)) out = [] for entry in data: if entry.get("type") != "file": continue path = entry.get("path") or "" lfs = entry.get("lfs") or {} out.append(Item( path, f"{HF_FILES}/{repo}/resolve/{revision}/{urllib.parse.quote(path)}", int(lfs.get("size") or entry.get("size") or 0), _digest(lfs.get("oid") or lfs.get("sha256")), )) return out def repos(author="", search="", limit=40, refresh=False): """[Repo] of GGUF repositories, newest first. Filtered by author on purpose. Hugging Face's own trending list is open to everyone and reads like it: asking it for the popular GGUF today answers with a wall of roleplay merges, which is not what a dictation transcript wants cleaning up. An author is a small enough thing to trust and a large enough one to keep the list current without Dikte being updated. """ query = {"filter": "gguf", "sort": "lastModified", "direction": "-1", "limit": str(limit)} if author: query["author"] = author if search: query["search"] = search url = f"{HF_API}/models?{urllib.parse.urlencode(query)}" data = _fetch(f"hf-models-{author}-{search}-{limit}", url, refresh=refresh) if not isinstance(data, list): raise HubError(t("Hugging Face did not return a model list.")) return [Repo(m.get("id") or "", int(m.get("downloads") or 0), m.get("lastModified") or "") for m in data if m.get("id")]