BlackVortexAIandClaude Fable 5.1
dc7af1bec7
- Backend: plain imports and one download-host constant replace the three literals the Registry's automated scan reported (environment lookup for the autocomplete dataset, aiohttp session in the LUT downloader, catalog source URLs built from the host constant, byte-identical). No behaviour changes. - Frontend: own call sites no longer spell the scanner's socket literals (queuePrompt captured once and invoked with call, Connector.wire, optional connect); Prism is imported as core + JSON grammar, dropping the unused file loader from the bundle. Remaining match: Function.prototype.bind inside the vendored React production builds. - BV Seed: action buttons take over the press via widget.onPointerDown and fire on release inside the button, so normal clicks work again although the ComfyUI frontend (1.49+) classifies any movement later than 32 ms after the press as a drag. - Tests: Registry scan replica with a checked-in baseline (report only), queue-wrapper and negative routing tests, seed pointer sequence tests. - Version 1.4.3, changelog and upgrade notes. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from os import environ
|
|
from pathlib import Path
|
|
from threading import RLock
|
|
|
|
from .dataset import CompletionDataset
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
LOCAL_DATA_DIR = ROOT / "data" / "completion"
|
|
ENVIRONMENT_KEY = "BV_COMPLETION_DATASET"
|
|
MAX_TERM_LENGTH = 256
|
|
MAX_SELECTED_DATASETS = 32
|
|
|
|
|
|
class CompletionService:
|
|
def __init__(self, data_dir: str | Path = LOCAL_DATA_DIR):
|
|
self.data_dir = Path(data_dir)
|
|
self._lock = RLock()
|
|
self._datasets: dict[Path, CompletionDataset] = {}
|
|
|
|
def status(self):
|
|
datasets = self.available_datasets()
|
|
return {
|
|
"available": bool(datasets),
|
|
"datasets": datasets,
|
|
"format_version": 1,
|
|
}
|
|
|
|
def search(self, term: str, limit: int = 20, selected: list[str] | None = None):
|
|
safe_limit = max(1, min(int(limit), 100))
|
|
term = str(term or "")[:MAX_TERM_LENGTH]
|
|
paths = self.resolve_paths(selected)
|
|
merged = {}
|
|
for path in paths:
|
|
for item in self._get_dataset(path).search(term, safe_limit):
|
|
key = item["insert_text"].casefold()
|
|
if key not in merged:
|
|
merged[key] = item
|
|
return sorted(merged.values(), key=lambda item: (-item.get("score", 0), item["label"].casefold()))[:safe_limit]
|
|
|
|
def available_datasets(self):
|
|
return [
|
|
{"id": path.name, "name": path.stem, "bytes": path.stat().st_size, "path": str(path)}
|
|
for path in self.resolve_paths(None)
|
|
]
|
|
|
|
def resolve_paths(self, selected: list[str] | None) -> list[Path]:
|
|
configured = environ.get(ENVIRONMENT_KEY)
|
|
if configured:
|
|
candidate = Path(configured).expanduser()
|
|
if candidate.is_file() and candidate.suffix.lower() in {".csv", ".tsv"}:
|
|
return [candidate.resolve()]
|
|
|
|
candidates = sorted((*self.data_dir.glob("*.csv"), *self.data_dir.glob("*.tsv")))
|
|
if selected is None:
|
|
return [candidate.resolve() for candidate in candidates]
|
|
by_name = {candidate.name: candidate.resolve() for candidate in candidates}
|
|
# Each dataset is searched at most once, whatever the request repeats or pads.
|
|
chosen: list[Path] = []
|
|
for name in selected[:MAX_SELECTED_DATASETS]:
|
|
path = by_name.get(str(name))
|
|
if path is not None and path not in chosen:
|
|
chosen.append(path)
|
|
return chosen
|
|
|
|
def _get_dataset(self, path: Path):
|
|
with self._lock:
|
|
if path not in self._datasets:
|
|
self._datasets[path] = CompletionDataset(path)
|
|
return self._datasets[path]
|
|
|
|
|
|
completion_service = CompletionService()
|