"""Example provider template for use as a starter kit. This minimal provider demonstrates the typical hooks a provider may implement: - `validate()` to assert it's usable - `search()` to return `SearchResult` items - `download()` to persist a sample file (useful for local tests) See `docs/provider_guide.md` for authoring guidance. """ from __future__ import annotations from pathlib import Path from typing import Any, Dict, List, Optional from ProviderCore.base import Provider, SearchResult class HelloProvider(Provider): """Very small example provider suitable as a template. - Table name: `hello` - Usage: `search-file -provider hello "query"` - Selecting a row and piping into `download-file` will call `download()`. """ URL = ("hello:",) URL_DOMAINS = () def validate(self) -> bool: # No configuration required; always available for testing/demo purposes. return True def search( self, query: str, limit: int = 50, filters: Optional[Dict[str, Any]] = None, **kwargs: Any, ) -> List[SearchResult]: q = (query or "").strip() results: List[SearchResult] = [] if not q or q in {"*", "all", "list"}: q = "example" # Emit up to `limit` tiny example results. n = min(max(1, int(limit)), 3) for i in range(1, n + 1): title = f"{q} sample {i}" path = f"https://example.org/{q}/{i}" sr = SearchResult( table="hello", title=title, path=path, detail="Example provider result", media_kind="file", columns=[("Example", "yes")], full_metadata={"example_index": i}, ) results.append(sr) return results[: max(0, int(limit))] def download(self, result: SearchResult, output_dir: Path) -> Optional[Path]: """Create a small text file to simulate a download. This keeps the example self-contained (no network access required) and makes it straightforward to test provider behavior with `pytest`. """ try: Path(output_dir).mkdir(parents=True, exist_ok=True) except Exception: pass title = str(getattr(result, "title", "hello") or "hello").strip() safe = "".join(c if c.isalnum() or c in ("-", "_", ".") else "_" for c in title) fname = f"{safe}.txt" if safe else "hello.txt" dest = Path(output_dir) / fname try: dest.write_text(f"Hello from HelloProvider\nsource: {result.path}\n", encoding="utf-8") return dest except Exception: return None def selector( self, selected_items: List[Any], *, ctx: Any, stage_is_last: bool = True, **_kwargs: Any, ) -> bool: """Present a simple details table when a HelloProvider row is selected. This demonstrates how providers can implement custom `@N` selection behavior by constructing a `ResultTable`, populating it with provider-specific rows, and instructing the CLI to show the table. """ if not stage_is_last: return False def _as_payload(item: Any) -> Dict[str, Any]: if isinstance(item, dict): return dict(item) try: if hasattr(item, "to_dict"): maybe = item.to_dict() if isinstance(maybe, dict): return maybe except Exception: pass payload: Dict[str, Any] = {} try: payload = { "title": getattr(item, "title", None), "path": getattr(item, "path", None), "table": getattr(item, "table", None), "annotations": getattr(item, "annotations", None), "media_kind": getattr(item, "media_kind", None), "full_metadata": getattr(item, "full_metadata", None), } except Exception: payload = {} return payload chosen: List[Dict[str, Any]] = [] for item in selected_items or []: payload = _as_payload(item) meta = payload.get("full_metadata") or {} if not isinstance(meta, dict): meta = {} idx = meta.get("example_index") if idx is None: continue title = str(payload.get("title") or payload.get("path") or "").strip() or f"hello-{idx}" chosen.append({"index": idx, "title": title, "path": payload.get("path")}) if not chosen: return False target = chosen[0] idx = target.get("index") title = target.get("title") or f"hello-{idx}" try: from SYS.result_table import ResultTable from SYS.rich_display import stdout_console except Exception: # If ResultTable isn't available, consider selection handled return True table = ResultTable(f"Hello Details: {title}").set_preserve_order(True) table.set_table("hello") try: table.set_table_metadata({"provider": "hello", "view": "details", "example_index": idx}) except Exception: pass table.set_source_command("download-file", []) results_payload: List[Dict[str, Any]] = [] for part in ("a", "b"): file_title = f"{title} - part {part}" file_path = f"{target.get('path')}/{part}" sr = SearchResult( table="hello", title=file_title, path=file_path, detail=f"Part {part}", media_kind="file", columns=[("Part", part)], full_metadata={"part": part, "example_index": idx}, ) table.add_result(sr) try: results_payload.append(sr.to_dict()) except Exception: results_payload.append({"table": sr.table, "title": sr.title, "path": sr.path}) try: ctx.set_last_result_table(table, results_payload) ctx.set_current_stage_table(table) except Exception: pass try: stdout_console().print() stdout_console().print(table) except Exception: pass return True