226 lines
7.6 KiB
Python
226 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = ["pyyaml"]
|
|
# ///
|
|
"""Layout and configuration for the wiki skill.
|
|
|
|
All runtime data lives under `workspace/wiki/`; only `config.yaml` is versioned.
|
|
|
|
The source id is the stable key — the catalog and the vectors hang off it, while a
|
|
URL or a path may change. Renaming an id is therefore an explicit invalidation of
|
|
that source's index, not a rename.
|
|
|
|
Scope precedence is `paths` (whitelist — what is not in it does not exist for the
|
|
index) then `include` (extension whitelist) then `exclude` (scalpel, wins over both).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
# workspace/skills/wiki/scripts/wiki_config.py -> parents[3] = workspace root.
|
|
WORKSPACE = Path(__file__).resolve().parents[3]
|
|
WIKI_DIR = WORKSPACE / "wiki"
|
|
DEFAULT_DB_PATH = WIKI_DIR / "index.sqlite"
|
|
DEFAULT_CONFIG_PATH = WIKI_DIR / "config.yaml"
|
|
REMOTE_DIR = WIKI_DIR / "remote"
|
|
LOCK_PATH = WIKI_DIR / ".sync.lock"
|
|
SYNC_LOG_PATH = WORKSPACE / "log" / "wiki_sync.log"
|
|
|
|
GIT_KIND = "git"
|
|
WORKSPACE_KIND = "workspace"
|
|
VALID_KINDS = (GIT_KIND, WORKSPACE_KIND)
|
|
|
|
|
|
def db_path() -> Path:
|
|
"""Index location, overridable with WIKI_DB for tests."""
|
|
return Path(os.environ.get("WIKI_DB", str(DEFAULT_DB_PATH)))
|
|
|
|
|
|
def config_path() -> Path:
|
|
"""Config location, overridable with WIKI_CONFIG for tests."""
|
|
return Path(os.environ.get("WIKI_CONFIG", str(DEFAULT_CONFIG_PATH)))
|
|
|
|
|
|
class ConfigError(ValueError):
|
|
"""Malformed or incomplete wiki/config.yaml."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EmbeddingConfig:
|
|
endpoint: str
|
|
model: str
|
|
dims: int
|
|
batch: int
|
|
keep_alive: int
|
|
query_prefix: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SourceConfig:
|
|
source_id: str
|
|
kind: str
|
|
url: str | None
|
|
paths: tuple[str, ...]
|
|
include: tuple[str, ...]
|
|
exclude: tuple[str, ...]
|
|
|
|
def covers(self, rel_path: str) -> bool:
|
|
"""True when a source-relative path belongs in the index."""
|
|
if not any(_matches(pattern, rel_path) for pattern in self.paths):
|
|
return False
|
|
name = rel_path.rsplit("/", 1)[-1]
|
|
if not any(_matches(pattern, name) for pattern in self.include):
|
|
return False
|
|
return not any(_matches(pattern, rel_path) for pattern in self.exclude)
|
|
|
|
def covers_dir(self, rel_dir: str) -> bool:
|
|
"""Cheap walk prune: could anything under this directory ever be covered?"""
|
|
if not rel_dir:
|
|
return True
|
|
probe = f"{rel_dir}/"
|
|
if any(_matches(pattern, probe) for pattern in self.exclude):
|
|
return False
|
|
return any(_prefix_could_match(pattern, probe) for pattern in self.paths)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WikiConfig:
|
|
embedding: EmbeddingConfig
|
|
sources: tuple[SourceConfig, ...]
|
|
|
|
def source(self, source_id: str) -> SourceConfig | None:
|
|
return next((s for s in self.sources if s.source_id == source_id), None)
|
|
|
|
|
|
def source_root(source: SourceConfig) -> Path:
|
|
"""Where a source's files live. Git clones are derived from the id, never configured."""
|
|
if source.kind == GIT_KIND:
|
|
return REMOTE_DIR / source.source_id
|
|
return WORKSPACE
|
|
|
|
|
|
def load_config(path: Path | None = None) -> WikiConfig:
|
|
path = path or config_path()
|
|
if not path.exists():
|
|
raise ConfigError(f"missing config: {path}")
|
|
try:
|
|
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
except yaml.YAMLError as exc:
|
|
raise ConfigError(f"unparseable config {path}: {exc}") from exc
|
|
if not isinstance(raw, dict):
|
|
raise ConfigError(f"config {path} must be a mapping")
|
|
return WikiConfig(
|
|
embedding=_parse_embedding(raw.get("embedding")),
|
|
sources=_parse_sources(raw.get("sources")),
|
|
)
|
|
|
|
|
|
def _as_mapping(raw: object, what: str) -> dict[str, object]:
|
|
if not isinstance(raw, dict):
|
|
raise ConfigError(f"{what} must be a mapping")
|
|
return {str(key): value for key, value in raw.items()}
|
|
|
|
|
|
def _parse_embedding(raw: object) -> EmbeddingConfig:
|
|
values = _as_mapping(raw, "`embedding`")
|
|
missing = [key for key in ("endpoint", "model", "dims") if key not in values]
|
|
if missing:
|
|
raise ConfigError(f"embedding is missing {', '.join(missing)}")
|
|
keep_alive = values.get("keep_alive", -1)
|
|
if not isinstance(keep_alive, int):
|
|
# Ollama rejects a string keep_alive of "-1" with HTTP 400.
|
|
raise ConfigError("embedding.keep_alive must be a number, not a string")
|
|
return EmbeddingConfig(
|
|
endpoint=str(values["endpoint"]).rstrip("/"),
|
|
model=str(values["model"]),
|
|
dims=int(str(values["dims"])),
|
|
batch=int(str(values.get("batch", 32))),
|
|
keep_alive=keep_alive,
|
|
query_prefix=str(values.get("query_prefix", "")),
|
|
)
|
|
|
|
|
|
def _parse_sources(raw: object) -> tuple[SourceConfig, ...]:
|
|
entries = _as_mapping(raw, "`sources`")
|
|
if not entries:
|
|
raise ConfigError("config needs a non-empty `sources` mapping")
|
|
sources = []
|
|
for source_id, raw_body in entries.items():
|
|
body = _as_mapping(raw_body, f"source {source_id}")
|
|
kind = str(body.get("kind"))
|
|
if kind not in VALID_KINDS:
|
|
raise ConfigError(f"source {source_id}: kind must be one of {VALID_KINDS}, got {kind!r}")
|
|
url = body.get("url")
|
|
if kind == GIT_KIND and not url:
|
|
raise ConfigError(f"source {source_id}: git sources need a url")
|
|
sources.append(
|
|
SourceConfig(
|
|
source_id=source_id,
|
|
kind=kind,
|
|
url=str(url) if url else None,
|
|
paths=_as_patterns(body.get("paths"), source_id, "paths"),
|
|
include=_as_patterns(body.get("include") or ["*.md"], source_id, "include"),
|
|
exclude=_as_patterns(body.get("exclude") or [], source_id, "exclude"),
|
|
)
|
|
)
|
|
return tuple(sources)
|
|
|
|
|
|
def _as_patterns(raw: object, source_id: str, key: str) -> tuple[str, ...]:
|
|
if raw is None:
|
|
raise ConfigError(f"source {source_id}: `{key}` is required")
|
|
if not isinstance(raw, list):
|
|
raise ConfigError(f"source {source_id}: `{key}` must be a list")
|
|
return tuple(str(item) for item in raw)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# glob matching
|
|
#
|
|
# fnmatch lets `*` cross a `/` and PurePath.match has no recursive `**` before
|
|
# Python 3.13, so the patterns are translated by hand.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SEGMENT_ANY = "[^/]*"
|
|
|
|
|
|
def _glob_to_regex(pattern: str) -> str:
|
|
parts = pattern.split("/")
|
|
out = []
|
|
for index, part in enumerate(parts):
|
|
is_last = index == len(parts) - 1
|
|
if part == "**":
|
|
out.append(".*" if is_last else "(?:[^/]+/)*")
|
|
else:
|
|
segment = re.escape(part).replace(r"\*", _SEGMENT_ANY).replace(r"\?", "[^/]")
|
|
out.append(segment if is_last else segment + "/")
|
|
return "^" + "".join(out) + "$"
|
|
|
|
|
|
def _matches(pattern: str, value: str) -> bool:
|
|
return re.match(_glob_to_regex(pattern), value) is not None
|
|
|
|
|
|
def _prefix_could_match(pattern: str, directory: str) -> bool:
|
|
"""True when `pattern` can still match something below `directory`."""
|
|
if pattern.startswith("**"):
|
|
return True
|
|
pattern_parts = pattern.split("/")
|
|
dir_parts = [part for part in directory.split("/") if part]
|
|
for depth, dir_part in enumerate(dir_parts):
|
|
if depth >= len(pattern_parts):
|
|
return False
|
|
pattern_part = pattern_parts[depth]
|
|
if pattern_part == "**":
|
|
return True
|
|
if not _matches(pattern_part, dir_part):
|
|
return False
|
|
return True
|