diff --git a/tools/fojin_bridge.py b/tools/fojin_bridge.py new file mode 100644 index 0000000..3b2b9af --- /dev/null +++ b/tools/fojin_bridge.py @@ -0,0 +1,130 @@ +""" +FoJin Data Bridge — connects buddha-skill to FoJin's Buddhist text platform. + +Two modes: +- API mode (default): calls fojin.app REST API, works for any user +- Local mode: direct database access, for FoJin developers only +""" + +import json +import os +from typing import Optional + +import requests + + +class FojinBridge: + """Bridge to FoJin Buddhist text platform.""" + + def __init__(self, mode: str = "api", base_url: str = "https://fojin.app"): + self.mode = mode + self.base_url = base_url.rstrip("/") + self.session = requests.Session() + self.session.headers.update({"Accept": "application/json"}) + + # ── Search ────────────────────────────────────────────── + + def search_texts(self, query: str, sources: Optional[str] = None, lang: Optional[str] = None, page: int = 1, size: int = 20) -> dict: + """Search Buddhist texts by keyword.""" + params = {"q": query, "page": page, "size": size} + if sources: + params["sources"] = sources + if lang: + params["lang"] = lang + return self._get("/api/search", params) + + def search_content(self, query: str, sources: Optional[str] = None, page: int = 1, size: int = 20) -> dict: + """Full-text content search with highlighting.""" + params = {"q": query, "page": page, "size": size} + if sources: + params["sources"] = sources + return self._get("/api/search/content", params) + + def semantic_search(self, query: str, top_k: int = 10) -> dict: + """Vector similarity search using pgvector embeddings.""" + params = {"q": query, "size": top_k} + return self._get("/api/search/semantic", params) + + # ── Texts ─────────────────────────────────────────────── + + def get_text(self, text_id: int) -> dict: + """Get text metadata by ID.""" + return self._get(f"/api/texts/{text_id}") + + def get_text_content(self, text_id: int, juan_num: int, lang: Optional[str] = None) -> dict: + """Get full content of a specific juan (scroll/fascicle).""" + params = {} + if lang: + params["lang"] = lang + return self._get(f"/api/texts/{text_id}/juans/{juan_num}", params) + + def get_text_juans(self, text_id: int) -> dict: + """List all juans for a text.""" + return self._get(f"/api/texts/{text_id}/juans") + + def lookup_cbeta_ids(self, ids: str) -> dict: + """Batch lookup CBETA IDs to internal IDs.""" + return self._get("/api/texts/lookup-cbeta", {"ids": ids}) + + def get_similar_passages(self, text_id: int, juan_num: int) -> dict: + """Find similar passages using pgvector similarity.""" + return self._get(f"/api/texts/{text_id}/juans/{juan_num}/similar") + + # ── Knowledge Graph ───────────────────────────────────── + + def search_kg_entities(self, query: str, entity_type: Optional[str] = None, limit: int = 20) -> dict: + """Search knowledge graph entities.""" + params = {"q": query, "limit": limit} + if entity_type: + params["entity_type"] = entity_type + return self._get("/api/kg/entities", params) + + def get_kg_entity(self, entity_id: int) -> dict: + """Get detailed entity info with relations.""" + return self._get(f"/api/kg/entities/{entity_id}") + + def get_kg_graph(self, entity_id: int, depth: int = 2, max_nodes: int = 150, predicates: Optional[str] = None) -> dict: + """Get entity's relationship graph.""" + params = {"depth": depth, "max_nodes": max_nodes} + if predicates: + params["predicates"] = predicates + return self._get(f"/api/kg/entities/{entity_id}/graph", params) + + # ── Dictionary ────────────────────────────────────────── + + def search_dictionary(self, query: str, lang: Optional[str] = None, source: Optional[str] = None, page: int = 1, size: int = 20) -> dict: + """Search Buddhist dictionaries.""" + params = {"q": query, "page": page, "size": size} + if lang: + params["lang"] = lang + if source: + params["source"] = source + return self._get("/api/dictionary/search", params) + + def search_dictionary_grouped(self, query: str) -> dict: + """Search dictionaries, results grouped by source.""" + return self._get("/api/dictionary/search/grouped", {"q": query}) + + # ── Helpers ────────────────────────────────────────────── + + def _get(self, path: str, params: Optional[dict] = None) -> dict: + """Make GET request to FoJin API.""" + url = f"{self.base_url}{path}" + resp = self.session.get(url, params=params, timeout=30) + resp.raise_for_status() + return resp.json() + + def test_connection(self) -> bool: + """Test if FoJin API is reachable.""" + try: + self._get("/api/stats") + return True + except Exception: + return False + + +def create_bridge() -> FojinBridge: + """Create a FojinBridge from environment variables.""" + mode = os.environ.get("FOJIN_MODE", "api") + url = os.environ.get("FOJIN_URL", "https://fojin.app") + return FojinBridge(mode=mode, base_url=url)