smith.infrastructure.template_source

  1"""Template source adapters — resolve files from bundled, local, or URL sources."""
  2
  3import importlib.resources
  4import os
  5import tarfile
  6import tempfile
  7import zipfile
  8from pathlib import Path
  9
 10import requests
 11
 12from smith.domain.ports import TemplateSourceError
 13from smith.domain.value_objects import FileSpec, TemplateSource
 14
 15AGENTIC_FILE_PATTERNS = [
 16    "AGENTS.md",
 17    ".opencode/agents/",
 18    ".opencode/knowledge/",
 19    ".opencode/skills/",
 20    ".opencode/tools/",
 21    ".templates/",
 22    ".flowr/",
 23]
 24
 25GITIGNORE_PATTERNS = ["AGENTS.md", ".opencode/", ".templates/", ".flowr/"]
 26
 27
 28def _is_agentic_path(path: Path) -> bool:
 29    """Return whether *path* belongs to an agentic file pattern."""
 30    path_str = str(path)
 31    if path_str == "AGENTS.md":
 32        return True
 33    return any(path_str.startswith(pattern) for pattern in AGENTIC_FILE_PATTERNS[1:])
 34
 35
 36def _collect_specs_from_directory(directory: Path) -> list[FileSpec]:
 37    """Walk *directory* and collect FileSpec for every file found."""
 38    specs: list[FileSpec] = []
 39    for root, _dirs, files in os.walk(directory):
 40        for f in files:
 41            full = Path(root) / f
 42            rel = full.relative_to(directory)
 43            specs.append(FileSpec(relative_path=rel, content=full.read_bytes()))
 44    return specs
 45
 46
 47class BundledTemplateSource:
 48    """Resolve templates shipped inside the ``smith.data`` package."""
 49
 50    def resolve(self) -> list[FileSpec]:
 51        """Return FileSpec list for every agentic file in the bundled data."""
 52        try:
 53            data_dir = importlib.resources.files("smith.data")
 54        except Exception as exc:
 55            raise TemplateSourceError(
 56                f"Failed to locate bundled template data: {exc}"
 57            ) from exc
 58
 59        if not hasattr(data_dir, "joinpath"):
 60            raise TemplateSourceError("Bundled template data directory not found")
 61
 62        data_path = Path(str(data_dir.joinpath()))
 63        if not data_path.is_dir():
 64            raise TemplateSourceError(
 65                f"Bundled template data directory not found: {data_path}"
 66            )
 67
 68        all_specs = _collect_specs_from_directory(data_path)
 69        specs = [s for s in all_specs if _is_agentic_path(s.relative_path)]
 70
 71        if not specs:
 72            raise TemplateSourceError("Bundled template data contains no agentic files")
 73
 74        return specs
 75
 76    def gitignore_patterns(self) -> list[str]:
 77        """Return the standard gitignore patterns for bundled templates."""
 78        return list(GITIGNORE_PATTERNS)
 79
 80
 81class LocalTemplateSource:
 82    """Resolve templates from a local directory on disk."""
 83
 84    def __init__(self, path: Path) -> None:
 85        """Initialise with the local template directory path."""
 86        self._path = path
 87
 88    def resolve(self) -> list[FileSpec]:
 89        """Return FileSpec list for every file in the local directory."""
 90        if not self._path.is_dir():
 91            raise TemplateSourceError(f"Template directory not found: {self._path}")
 92        return _collect_specs_from_directory(self._path)
 93
 94    def gitignore_patterns(self) -> list[str]:
 95        """Return gitignore patterns derived from the local directory contents."""
 96        patterns: list[str] = []
 97        for item in sorted(self._path.iterdir()):
 98            name = item.name
 99            if item.is_dir():
100                patterns.append(f"{name}/")
101            else:
102                patterns.append(name)
103        return patterns
104
105
106class UrlTemplateSource:
107    """Resolve templates from a remote tar.gz or zip archive."""
108
109    def __init__(self, url: str) -> None:
110        """Initialise with the URL of the template archive."""
111        self._url = url
112
113    def resolve(self) -> list[FileSpec]:
114        """Download, extract, and return FileSpec list for agentic files."""
115        try:
116            response = requests.get(self._url, timeout=30)
117            response.raise_for_status()
118        except requests.RequestException as exc:
119            raise TemplateSourceError(
120                f"Failed to download template from {self._url}: {exc}"
121            ) from exc
122
123        tmp_dir = tempfile.mkdtemp(prefix="smith_url_")
124        try:
125            try:
126                if self._url.endswith(".zip"):
127                    self._extract_zip(response.content, tmp_dir)
128                else:
129                    self._extract_tar(response.content, tmp_dir)
130            except (tarfile.TarError, zipfile.BadZipFile, OSError) as exc:
131                raise TemplateSourceError(
132                    f"Failed to extract template archive from {self._url}: {exc}"
133                ) from exc
134
135            all_specs = _collect_specs_from_directory(Path(tmp_dir))
136        finally:
137            import shutil
138
139            shutil.rmtree(tmp_dir, ignore_errors=True)
140
141        specs = [s for s in all_specs if _is_agentic_path(s.relative_path)]
142
143        if not specs:
144            raise TemplateSourceError(
145                f"Template archive from {self._url} contains no agentic files"
146            )
147
148        return specs
149
150    def gitignore_patterns(self) -> list[str]:
151        """Return the standard gitignore patterns for URL-sourced templates."""
152        return list(GITIGNORE_PATTERNS)
153
154    @staticmethod
155    def _extract_tar(content: bytes, target_dir: str) -> None:
156        """Extract a tar.gz archive, stripping the top-level directory."""
157        import io
158
159        with tarfile.open(fileobj=io.BytesIO(content), mode="r:gz") as tar:
160            members = tar.getmembers()
161            if not members:
162                raise TemplateSourceError("Tar archive is empty")
163
164            root_dir = members[0].name.split("/")[0]
165            for member in members:
166                member_path = Path(member.name)
167                if member_path.parts[0] == root_dir:
168                    member.name = str(Path(*member_path.parts[1:]))
169                if member.name:
170                    tar.extract(member, target_dir, filter="data")
171
172    @staticmethod
173    def _extract_zip(content: bytes, target_dir: str) -> None:
174        """Extract a zip archive, stripping the top-level directory."""
175        import io
176
177        with zipfile.ZipFile(io.BytesIO(content)) as zf:
178            names = zf.namelist()
179            if not names:
180                raise TemplateSourceError("Zip archive is empty")
181
182            root_dir = names[0].split("/")[0]
183            for name in names:
184                if name == root_dir + "/":
185                    continue
186                if name.startswith(root_dir + "/"):
187                    zf.extract(name, target_dir)
188
189
190class TemplateSourceAdapter:
191    """Dispatch to the correct TemplateSourcePort implementation based on kind."""
192
193    def __init__(self, source: TemplateSource) -> None:
194        """Initialise with the TemplateSource value object."""
195        self._source = source
196
197    def resolve(self) -> list[FileSpec]:
198        """Resolve template files via the appropriate source strategy."""
199        if self._source.kind == "bundled":
200            return BundledTemplateSource().resolve()
201        if self._source.kind == "local":
202            return LocalTemplateSource(Path(self._source.location)).resolve()
203        if self._source.kind == "url":
204            return UrlTemplateSource(self._source.location).resolve()
205        raise TemplateSourceError(f"Unknown template source kind: {self._source.kind}")
206
207    def gitignore_patterns(self) -> list[str]:
208        """Return gitignore patterns via the appropriate source strategy."""
209        if self._source.kind == "bundled":
210            return BundledTemplateSource().gitignore_patterns()
211        if self._source.kind == "local":
212            return LocalTemplateSource(Path(self._source.location)).gitignore_patterns()
213        if self._source.kind == "url":
214            return UrlTemplateSource(self._source.location).gitignore_patterns()
215        raise TemplateSourceError(f"Unknown template source kind: {self._source.kind}")