125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
import os
|
|
import httpx
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
|
|
class VndbClient:
|
|
"""
|
|
Minimal safe VNDB client for Telegram bots
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.base_url = os.getenv("VNDB_API_URL", "https://api.vndb.org/kana")
|
|
self.token = os.getenv("VNDB_TOKEN")
|
|
|
|
self.headers = {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
if self.token:
|
|
self.headers["Authorization"] = f"Token {self.token}"
|
|
|
|
# =========================
|
|
# CORE REQUEST
|
|
# =========================
|
|
async def _request(self, endpoint: str, payload: dict):
|
|
url = f"{self.base_url}{endpoint}"
|
|
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
r = await client.post(url, json=payload, headers=self.headers)
|
|
|
|
# debug-friendly
|
|
if r.status_code >= 400:
|
|
print("VNDB ERROR:", r.status_code, r.text)
|
|
print("PAYLOAD:", payload)
|
|
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
# =========================
|
|
# SAFE SEARCH WRAPPER
|
|
# =========================
|
|
def _safe_search(self, query: str):
|
|
if not query:
|
|
return None
|
|
return ["search", "=", query.strip()]
|
|
|
|
# =========================
|
|
# VN SEARCH
|
|
# =========================
|
|
async def search_vn(self, query: str, limit: int = 10):
|
|
filters = self._safe_search(query)
|
|
if not filters:
|
|
return {"results": []}
|
|
|
|
return await self._request("/vn", {
|
|
"filters": filters,
|
|
"fields": "id,title,original,released,rating,votecount",
|
|
"results": limit
|
|
})
|
|
|
|
# =========================
|
|
# CHARACTER SEARCH
|
|
# =========================
|
|
async def search_character(self, query: str, limit: int = 10):
|
|
filters = self._safe_search(query)
|
|
if not filters:
|
|
return {"results": []}
|
|
|
|
return await self._request("/character", {
|
|
"filters": filters,
|
|
"fields": "id,name,original",
|
|
"results": limit
|
|
})
|
|
|
|
# =========================
|
|
# RELEASE SEARCH
|
|
# =========================
|
|
async def search_release(self, query: str, limit: int = 10):
|
|
filters = self._safe_search(query)
|
|
if not filters:
|
|
return {"results": []}
|
|
|
|
return await self._request("/release", {
|
|
"filters": filters,
|
|
"fields": "id,title,original,released",
|
|
"results": limit
|
|
})
|
|
|
|
# =========================
|
|
# DETAIL VN
|
|
# =========================
|
|
async def vn_detail(self, vn_id: str):
|
|
return await self._request("/vn", {
|
|
"filters": ["id", "=", vn_id],
|
|
"fields": "id,title,original,released,rating,votecount,description,length,developer",
|
|
"results": 1
|
|
})
|
|
|
|
# =========================
|
|
# DETAIL CHARACTER
|
|
# =========================
|
|
async def character_detail(self, char_id: str):
|
|
return await self._request("/character", {
|
|
"filters": ["id", "=", char_id],
|
|
"fields": "id,name,original,gender,bloodtype",
|
|
"results": 1
|
|
})
|
|
|
|
# =========================
|
|
# DETAIL RELEASE
|
|
# =========================
|
|
async def release_detail(self, rel_id: str):
|
|
return await self._request("/release", {
|
|
"filters": ["id", "=", rel_id],
|
|
"fields": "id,title,original,released,platform,type,language,description",
|
|
"results": 1
|
|
})
|
|
|
|
# =========================
|
|
# STATS
|
|
# =========================
|
|
async def stats(self):
|
|
return await self._request("/stats", {}) |