497 lines
15 KiB
Python
497 lines
15 KiB
Python
"""
|
|
VNDB API Client
|
|
Handles all interactions with the VNDB API
|
|
"""
|
|
import httpx
|
|
import json
|
|
from typing import Dict, List, Any, Optional
|
|
from enum import Enum
|
|
|
|
|
|
class VndbEndpoint(Enum):
|
|
"""VNDB API Endpoints"""
|
|
SCHEMA = "/schema"
|
|
STATS = "/stats"
|
|
USER = "/user"
|
|
AUTHINFO = "/authinfo"
|
|
VN = "/vn"
|
|
RELEASE = "/release"
|
|
CHARACTER = "/character"
|
|
STAFF = "/staff"
|
|
PRODUCER = "/producer"
|
|
TAG = "/tag"
|
|
TRAIT = "/trait"
|
|
QUOTE = "/quote"
|
|
ULIST = "/ulist"
|
|
RLIST = "/rlist"
|
|
ULIST_LABELS = "/ulist_labels"
|
|
|
|
|
|
class VndbClient:
|
|
"""Client for interacting with VNDB API"""
|
|
|
|
BASE_URL = "https://api.vndb.org/kana"
|
|
SANDBOX_URL = "https://beta.vndb.org/api/kana"
|
|
|
|
def __init__(self, token: Optional[str] = None, use_sandbox: bool = False):
|
|
"""
|
|
Initialize VNDB client
|
|
|
|
Args:
|
|
token: Optional API token for authenticated requests
|
|
use_sandbox: Whether to use sandbox endpoint
|
|
"""
|
|
self.token = token
|
|
self.base_url = self.SANDBOX_URL if use_sandbox else self.BASE_URL
|
|
self.headers = {
|
|
"Content-Type": "application/json",
|
|
}
|
|
if token:
|
|
self.headers["Authorization"] = f"Token {token}"
|
|
|
|
async def _request(
|
|
self,
|
|
endpoint: str,
|
|
method: str = "GET",
|
|
data: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Make HTTP request to VNDB API
|
|
|
|
Args:
|
|
endpoint: API endpoint
|
|
method: HTTP method (GET, POST, PATCH, DELETE)
|
|
data: Request body data
|
|
|
|
Returns:
|
|
Response JSON
|
|
|
|
Raises:
|
|
httpx.HTTPError: On HTTP errors
|
|
json.JSONDecodeError: On invalid JSON response
|
|
"""
|
|
url = f"{self.base_url}{endpoint}"
|
|
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
response = await client.request(
|
|
method,
|
|
url,
|
|
headers=self.headers,
|
|
content=json.dumps(data) if data else None,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
# Simple Requests
|
|
|
|
async def get_schema(self) -> Dict[str, Any]:
|
|
"""Get API schema with metadata"""
|
|
return await self._request(VndbEndpoint.SCHEMA.value)
|
|
|
|
async def get_stats(self) -> Dict[str, Any]:
|
|
"""Get database statistics"""
|
|
return await self._request(VndbEndpoint.STATS.value)
|
|
|
|
async def get_user(self, queries: List[str], fields: Optional[List[str]] = None) -> Dict[str, Any]:
|
|
"""
|
|
Lookup users by ID or username
|
|
|
|
Args:
|
|
queries: List of user IDs or usernames to lookup
|
|
fields: List of fields to retrieve (lengthvotes, lengthvotes_sum)
|
|
|
|
Returns:
|
|
User information
|
|
"""
|
|
params = {"q": queries}
|
|
if fields:
|
|
params["fields"] = fields
|
|
|
|
# Build query string
|
|
query_parts = [f"q={q}" for q in queries]
|
|
if fields:
|
|
query_parts.append(f"fields={','.join(fields)}")
|
|
query_string = "&".join(query_parts)
|
|
|
|
url = f"{self.base_url}{VndbEndpoint.USER.value}?{query_string}"
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
response = await client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_authinfo(self) -> Dict[str, Any]:
|
|
"""Get authenticated user information"""
|
|
if not self.token:
|
|
raise ValueError("Token required for authinfo")
|
|
return await self._request(VndbEndpoint.AUTHINFO.value)
|
|
|
|
# Database Querying
|
|
|
|
async def query_vn(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
count: bool = False,
|
|
user: Optional[str] = None,
|
|
compact_filters: bool = False,
|
|
normalized_filters: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Query visual novels
|
|
|
|
Args:
|
|
filters: Filter conditions
|
|
fields: Fields to retrieve
|
|
sort: Sort field (id, title, released, rating, votecount, searchrank)
|
|
reverse: Sort in descending order
|
|
results: Number of results per page (max 100)
|
|
page: Page number starting from 1
|
|
count: Include total count
|
|
user: User ID for user-specific filters
|
|
compact_filters: Include compact filter representation
|
|
normalized_filters: Include normalized filter representation
|
|
|
|
Returns:
|
|
Query results
|
|
"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
"compact_filters": compact_filters,
|
|
"normalized_filters": normalized_filters,
|
|
}
|
|
if user:
|
|
data["user"] = user
|
|
|
|
return await self._request(VndbEndpoint.VN.value, "POST", data)
|
|
|
|
async def query_release(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
count: bool = False,
|
|
user: Optional[str] = None,
|
|
compact_filters: bool = False,
|
|
normalized_filters: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Query releases"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
"compact_filters": compact_filters,
|
|
"normalized_filters": normalized_filters,
|
|
}
|
|
if user:
|
|
data["user"] = user
|
|
|
|
return await self._request(VndbEndpoint.RELEASE.value, "POST", data)
|
|
|
|
async def query_character(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
count: bool = False,
|
|
compact_filters: bool = False,
|
|
normalized_filters: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Query characters"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
"compact_filters": compact_filters,
|
|
"normalized_filters": normalized_filters,
|
|
}
|
|
|
|
return await self._request(VndbEndpoint.CHARACTER.value, "POST", data)
|
|
|
|
async def query_staff(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
count: bool = False,
|
|
compact_filters: bool = False,
|
|
normalized_filters: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Query staff"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
"compact_filters": compact_filters,
|
|
"normalized_filters": normalized_filters,
|
|
}
|
|
|
|
return await self._request(VndbEndpoint.STAFF.value, "POST", data)
|
|
|
|
async def query_producer(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
count: bool = False,
|
|
compact_filters: bool = False,
|
|
normalized_filters: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Query producers"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
"compact_filters": compact_filters,
|
|
"normalized_filters": normalized_filters,
|
|
}
|
|
|
|
return await self._request(VndbEndpoint.PRODUCER.value, "POST", data)
|
|
|
|
async def query_tag(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
count: bool = False,
|
|
compact_filters: bool = False,
|
|
normalized_filters: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Query tags"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
"compact_filters": compact_filters,
|
|
"normalized_filters": normalized_filters,
|
|
}
|
|
|
|
return await self._request(VndbEndpoint.TAG.value, "POST", data)
|
|
|
|
async def query_trait(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
count: bool = False,
|
|
compact_filters: bool = False,
|
|
normalized_filters: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Query traits"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
"compact_filters": compact_filters,
|
|
"normalized_filters": normalized_filters,
|
|
}
|
|
|
|
return await self._request(VndbEndpoint.TRAIT.value, "POST", data)
|
|
|
|
async def query_quote(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
count: bool = False,
|
|
compact_filters: bool = False,
|
|
normalized_filters: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Query quotes"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
"compact_filters": compact_filters,
|
|
"normalized_filters": normalized_filters,
|
|
}
|
|
|
|
return await self._request(VndbEndpoint.QUOTE.value, "POST", data)
|
|
|
|
# List Management
|
|
|
|
async def query_ulist(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
user: Optional[str] = None,
|
|
count: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Query user visual novel list"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
}
|
|
if user:
|
|
data["user"] = user
|
|
|
|
return await self._request(VndbEndpoint.ULIST.value, "POST", data)
|
|
|
|
async def query_rlist(
|
|
self,
|
|
filters: Optional[List[Any]] = None,
|
|
fields: Optional[List[str]] = None,
|
|
sort: str = "id",
|
|
reverse: bool = False,
|
|
results: int = 10,
|
|
page: int = 1,
|
|
user: Optional[str] = None,
|
|
count: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Query user release list"""
|
|
data = {
|
|
"filters": filters or [],
|
|
"fields": ",".join(fields) if fields else "",
|
|
"sort": sort,
|
|
"reverse": reverse,
|
|
"results": results,
|
|
"page": page,
|
|
"count": count,
|
|
}
|
|
if user:
|
|
data["user"] = user
|
|
|
|
return await self._request(VndbEndpoint.RLIST.value, "POST", data)
|
|
|
|
async def get_ulist_labels(self, user: Optional[str] = None) -> Dict[str, Any]:
|
|
"""Get user list labels"""
|
|
url = f"{self.base_url}{VndbEndpoint.ULIST_LABELS.value}"
|
|
if user:
|
|
url += f"?user={user}"
|
|
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
response = await client.get(url, headers=self.headers)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def add_to_ulist(
|
|
self,
|
|
vn_id: str,
|
|
status: Optional[str] = None,
|
|
notes: Optional[str] = None,
|
|
labels: Optional[List[str]] = None,
|
|
voted: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Add or update visual novel in user list"""
|
|
if not self.token:
|
|
raise ValueError("Token required for list operations")
|
|
|
|
data = {"id": vn_id}
|
|
if status:
|
|
data["status"] = status
|
|
if notes:
|
|
data["notes"] = notes
|
|
if labels:
|
|
data["labels"] = labels
|
|
if voted is not None:
|
|
data["voted"] = voted
|
|
|
|
return await self._request(
|
|
f"{VndbEndpoint.ULIST.value}/{vn_id}",
|
|
"PATCH",
|
|
data
|
|
)
|
|
|
|
async def add_to_rlist(
|
|
self,
|
|
release_id: str,
|
|
status: Optional[str] = None,
|
|
notes: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Add or update release in user list"""
|
|
if not self.token:
|
|
raise ValueError("Token required for list operations")
|
|
|
|
data = {"id": release_id}
|
|
if status:
|
|
data["status"] = status
|
|
if notes:
|
|
data["notes"] = notes
|
|
|
|
return await self._request(
|
|
f"{VndbEndpoint.RLIST.value}/{release_id}",
|
|
"PATCH",
|
|
data
|
|
)
|
|
|
|
async def remove_from_ulist(self, vn_id: str) -> None:
|
|
"""Remove visual novel from user list"""
|
|
if not self.token:
|
|
raise ValueError("Token required for list operations")
|
|
|
|
await self._request(f"{VndbEndpoint.ULIST.value}/{vn_id}", "DELETE")
|
|
|
|
async def remove_from_rlist(self, release_id: str) -> None:
|
|
"""Remove release from user list"""
|
|
if not self.token:
|
|
raise ValueError("Token required for list operations")
|
|
|
|
await self._request(f"{VndbEndpoint.RLIST.value}/{release_id}", "DELETE")
|