devplacepy/controller.py at acccdd33f70e294ae8968bdccd0ec55c66da76d1

 # retoor <retoor@molodetz.nl>
 from __future__ import annotations
 import json
 import logging
 from typing import Any
 import httpx
 from ..config import Settings
 from ..errors import NetworkError, ToolInputError, UpstreamError
 logger = logging.getLogger("devii.rsearch")
 USER_AGENT = "devii/0.1"
 def _flag(value: Any) -> str:
     if isinstance(value, bool):
         return "true" if value else "false"
     return "true" if str(value).strip().lower() in ("1", "true", "yes", "on") else "false"
 class RsearchController:
     def __init__(self, settings: Settings) -> None:
         self._settings = settings
     async def dispatch(self, name: str, arguments: dict[str, Any]) -> str:
         if not self._settings.rsearch_enabled:
             raise ToolInputError(
                 "Web search tools are disabled by the administrator; use platform tools instead."
             )
         if name == "rsearch":
             return await self._search(arguments)
         if name == "rsearch_answer":
             return await self._answer(arguments)
         if name == "rsearch_chat":
             return await self._chat(arguments)
         if name == "rsearch_describe_image":
             return await self._describe(arguments)
         raise ToolInputError(f"Unknown rsearch tool: {name}")
     async def _request(self, path: str, params: dict[str, Any]) -> dict[str, Any]:
         headers = {"User-Agent": USER_AGENT, "Accept": "application/json"}
         timeout = httpx.Timeout(self._settings.rsearch_timeout_seconds, connect=30.0)
         try:
             async with httpx.AsyncClient(
                 base_url=self._settings.rsearch_url,
                 headers=headers,
                 follow_redirects=True,
                 timeout=timeout,
             ) as client:
                 response = await client.get(path, params=params)
         except httpx.TimeoutException as exc:
             raise NetworkError(f"rsearch timed out calling {path}", path=path) from exc
         except httpx.HTTPError as exc:
             raise NetworkError(f"rsearch request failed: {exc}", path=path) from exc
         if response.status_code >= 400:
             raise UpstreamError(
                 f"rsearch returned {response.status_code} for {path}",
                 status=response.status_code,
                 path=path,
             )
         try:
             return response.json()
         except ValueError as exc:
             raise UpstreamError("rsearch returned a non-JSON response", path=path) from exc
     def _require_text(self, arguments: dict[str, Any], key: str) -> str:
         value = str(arguments.get(key, "")).strip()
         if not value:
             raise ToolInputError(f"rsearch requires a non-empty '{key}'.")
         return value
     def _count(self, arguments: dict[str, Any]) -> int:
         try:
             count = int(arguments.get("count", 10))
         except (TypeError, ValueError):
             count = 10
         return max(1, min(count, 100))
     async def _search(self, arguments: dict[str, Any]) -> str:
         query = self._require_text(arguments, "query")
         result_type = str(arguments.get("type", "web")).strip().lower()
         if result_type not in ("web", "images"):
             result_type = "web"
         params: dict[str, Any] = {
             "query": query,
             "count": self._count(arguments),
             "content": _flag(arguments.get("content")),
             "deep": _flag(arguments.get("deep")),
         }
         if result_type == "images":
             params["type"] = "images"
         data = await self._request("/search", params)
         results = data.get("results") or []
         compact = [
             {
                 "title": item.get("title"),
                 "url": item.get("url"),
                 "description": item.get("description"),
                 "source": item.get("source"),
                 "content": item.get("content"),
             }
             for item in results
         ]
         return json.dumps({
             "status": "success",
             "query": data.get("query", query),
             "type": result_type,
             "source": data.get("source"),
             "count": len(compact),
             "results": compact,
         }, ensure_ascii=False)
     async def _answer(self, arguments: dict[str, Any]) -> str:
         query = self._require_text(arguments, "query")
         params = {
             "query": query,
             "ai": "true",
             "content": _flag(arguments.get("content")),
             "count": self._count(arguments),
         }
         data = await self._request("/search", params)
         results = data.get("results") or []
         return json.dumps({
             "status": "success",
             "query": query,
             "answer": data.get("ai_response"),
             "answer_error": data.get("ai_error"),
             "sources": [
                 {"title": item.get("title"), "url": item.get("url")}
                 for item in results
             ],
         }, ensure_ascii=False)
     async def _chat(self, arguments: dict[str, Any]) -> str:
         prompt = self._require_text(arguments, "prompt")
         params: dict[str, Any] = {"prompt": prompt, "json": _flag(arguments.get("json"))}
         system = str(arguments.get("system", "")).strip()
         if system:
             params["system"] = system
         data = await self._request("/chat", params)
         return json.dumps({
             "status": "success",
             "response": data.get("response"),
             "json_mode": data.get("json_mode", False),
         }, ensure_ascii=False)
     async def _describe(self, arguments: dict[str, Any]) -> str:
         url = self._require_text(arguments, "url")
         data = await self._request("/describe", {"url": url})
         return json.dumps({
             "status": "success",
             "url": url,
             "mime_type": data.get("mime_type"),
             "description": data.get("description"),
         }, ensure_ascii=False)