"""
Search tools for ThePornDB MCP Service.
Provides search functionality for scenes, movies, and JAV content.
"""
import logging
from datetime import datetime
from mcp.server.fastmcp import Context
from mcp.server.session import ServerSession
from mcp.types import CallToolResult, TextContent
from ..models import SearchResponse
from ..api import ThePornDBAPI
logger = logging.getLogger(__name__)
async def search_scenes(
search_term: str,
year: int | None = None,
ctx: Context[ServerSession, None] = None
):
"""
Search for adult video scenes by term and optional year.
Args:
search_term: Search query (scene title, performer name, site name)
year: Optional filter by release year (1900-current year)
ctx: MCP context (injected automatically)
Returns:
SearchResponse with paginated scene results
Raises:
CallToolResult: With error if validation fails
"""
# Input validation
if not search_term or not search_term.strip():
return CallToolResult(
content=[TextContent(
type="text",
text="Error: Search term cannot be empty. Please provide a valid search term."
)],
isError=True
)
# Year validation
if year is not None:
current_year = datetime.now().year
if year < 1900 or year > current_year:
return CallToolResult(
content=[TextContent(
type="text",
text=f"Error: Year must be between 1900 and {current_year}"
)],
isError=True
)
# Get API client from context
if ctx and hasattr(ctx, 'request_context'):
api = ctx.request_context.lifespan_context.api
else:
# Fallback for testing without context
return CallToolResult(
content=[TextContent(
type="text",
text="Error: Server context not available"
)],
isError=True
)
await ctx.info(f"Searching scenes for '{search_term}'" + (f" in {year}" if year else ""))
try:
# Call API
result = api.search_data(search_term, "scenes", year)
if result is None:
return CallToolResult(
content=[TextContent(
type="text",
text="Error: Search failed. Please try again."
)],
isError=True
)
await ctx.info(f"Found {len(result['data'])} scenes")
return result
except Exception as e:
logger.error(f"Error searching scenes: {e}")
return CallToolResult(
content=[TextContent(
type="text",
text=f"Error: Failed to search scenes - {str(e)}"
)],
isError=True
)
async def search_movies(
search_term: str,
year: int | None = None,
ctx: Context[ServerSession, None] = None
):
"""
Search for full-length movies by term and optional year.
Args:
search_term: Search query (movie title, series, studio)
year: Optional filter by release year (1900-current year)
ctx: MCP context (injected automatically)
Returns:
SearchResponse with paginated movie results
Raises:
CallToolResult: With error if validation fails
"""
# Input validation
if not search_term or not search_term.strip():
return CallToolResult(
content=[TextContent(
type="text",
text="Error: Search term cannot be empty. Please provide a valid search term."
)],
isError=True
)
# Year validation
if year is not None:
current_year = datetime.now().year
if year < 1900 or year > current_year:
return CallToolResult(
content=[TextContent(
type="text",
text=f"Error: Year must be between 1900 and {current_year}"
)],
isError=True
)
# Get API client from context
if ctx and hasattr(ctx, 'request_context'):
api = ctx.request_context.lifespan_context.api
else:
return CallToolResult(
content=[TextContent(
type="text",
text="Error: Server context not available"
)],
isError=True
)
await ctx.info(f"Searching movies for '{search_term}'" + (f" in {year}" if year else ""))
try:
# Call API
result = api.search_data(search_term, "movies", year)
if result is None:
return CallToolResult(
content=[TextContent(
type="text",
text="Error: Search failed. Please try again."
)],
isError=True
)
await ctx.info(f"Found {len(result['data'])} movies")
return result
except Exception as e:
logger.error(f"Error searching movies: {e}")
return CallToolResult(
content=[TextContent(
type="text",
text=f"Error: Failed to search movies - {str(e)}"
)],
isError=True
)
async def search_jav(
search_term: str,
year: int | None = None,
ctx: Context[ServerSession, None] = None
):
"""
Search for Japanese Adult Video content by term and optional year.
Args:
search_term: Search query (JAV title, code, performer name)
year: Optional filter by release year (1900-current year)
ctx: MCP context (injected automatically)
Returns:
SearchResponse with paginated JAV results
Raises:
CallToolResult: With error if validation fails
"""
# Input validation
if not search_term or not search_term.strip():
return CallToolResult(
content=[TextContent(
type="text",
text="Error: Search term cannot be empty. Please provide a valid search term."
)],
isError=True
)
# Year validation
if year is not None:
current_year = datetime.now().year
if year < 1900 or year > current_year:
return CallToolResult(
content=[TextContent(
type="text",
text=f"Error: Year must be between 1900 and {current_year}"
)],
isError=True
)
# Get API client from context
if ctx and hasattr(ctx, 'request_context'):
api = ctx.request_context.lifespan_context.api
else:
return CallToolResult(
content=[TextContent(
type="text",
text="Error: Server context not available"
)],
isError=True
)
await ctx.info(f"Searching JAV content for '{search_term}'" + (f" in {year}" if year else ""))
try:
# Call API
result = api.search_data(search_term, "jav", year)
if result is None:
return CallToolResult(
content=[TextContent(
type="text",
text="Error: Search failed. Please try again."
)],
isError=True
)
await ctx.info(f"Found {len(result['data'])} JAV results")
return result
except Exception as e:
logger.error(f"Error searching JAV content: {e}")
return CallToolResult(
content=[TextContent(
type="text",
text=f"Error: Failed to search JAV content - {str(e)}"
)],
isError=True
)
