feat: infrance

traderai/starcitizen_wiki_client.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import quote
+
+import httpx
+
+
+class StarCitizenWikiError(RuntimeError):
+    pass
+
+
+class StarCitizenWikiClient:
+    def __init__(
+        self,
+        base_url: str = "https://starcitizen.tools",
+        api_base_url: str = "https://api.star-citizen.wiki",
+    ) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.api_base_url = api_base_url.rstrip("/")
+
+    async def search_pages(self, query: str, limit: int = 5) -> list[dict[str, Any]]:
+        body = await self._get_json(
+            f"{self.base_url}/api.php",
+            params={
+                "action": "query",
+                "generator": "prefixsearch",
+                "gpssearch": query,
+                "gpslimit": max(1, min(limit, 10)),
+                "prop": "description|pageimages|extracts",
+                "exintro": 1,
+                "explaintext": 1,
+                "exchars": 320,
+                "piprop": "thumbnail",
+                "pithumbsize": 240,
+                "format": "json",
+            },
+        )
+        pages = body.get("query", {}).get("pages", {})
+        ordered = sorted(
+            (item for item in pages.values() if isinstance(item, dict)),
+            key=lambda item: int(item.get("index") or 0),
+        )
+        return [
+            {
+                "pageid": item.get("pageid"),
+                "title": item.get("title"),
+                "description": item.get("description"),
+                "extract": item.get("extract"),
+                "thumbnail": (item.get("thumbnail") or {}).get("source"),
+                "url": f"{self.base_url}/{quote(str(item.get('title') or '').replace(' ', '_'), safe=':/_')}",
+            }
+            for item in ordered
+            if item.get("title")
+        ]
+
+    async def get_page_summary(self, title: str | None = None, pageid: int | None = None, chars: int = 700) -> dict[str, Any] | None:
+        params: dict[str, Any] = {
+            "action": "query",
+            "prop": "extracts|description|pageimages",
+            "exintro": 1,
+            "explaintext": 1,
+            "exchars": max(120, min(chars, 1200)),
+            "piprop": "thumbnail",
+            "pithumbsize": 320,
+            "format": "json",
+        }
+        if pageid is not None:
+            params["pageids"] = pageid
+        elif title:
+            params["titles"] = title
+        else:
+            raise StarCitizenWikiError("title or pageid is required")
+
+        body = await self._get_json(f"{self.base_url}/api.php", params=params)
+        pages = body.get("query", {}).get("pages", {})
+        for item in pages.values():
+            if isinstance(item, dict) and item.get("pageid") and item.get("title"):
+                return {
+                    "pageid": item.get("pageid"),
+                    "title": item.get("title"),
+                    "description": item.get("description"),
+                    "extract": item.get("extract"),
+                    "thumbnail": (item.get("thumbnail") or {}).get("source"),
+                    "url": f"{self.base_url}/{quote(str(item.get('title') or '').replace(' ', '_'), safe=':/_')}",
+                }
+        return None
+
+    async def search_verse(self, query: str) -> list[dict[str, Any]]:
+        body = await self._get_json(
+            f"{self.api_base_url}/api/search",
+            params={"filter[query]": query},
+        )
+        data = body.get("data")
+        return data if isinstance(data, list) else []
+
+    async def get_vehicle(self, slug: str) -> dict[str, Any]:
+        body = await self._get_json(f"{self.api_base_url}/api/vehicles/{slug.strip('/')}")
+        data = body.get("data")
+        if not isinstance(data, dict):
+            raise StarCitizenWikiError(f"Vehicle response for {slug} was not an object.")
+        return data
+
+    async def _get_json(self, url: str, params: dict[str, Any] | None = None) -> Any:
+        async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
+            response = await client.get(url, params=params, headers={"Accept": "application/json"})
+        try:
+            body = response.json()
+        except ValueError as exc:
+            raise StarCitizenWikiError(f"Star Citizen Wiki returned non-JSON response: HTTP {response.status_code}") from exc
+        if response.status_code >= 400:
+            raise StarCitizenWikiError(f"Star Citizen Wiki HTTP {response.status_code}: {body}")
+        return body