Compare commits
11
Commits
05a2a5010c
..
master
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
627ab3aec6 | ||
|
|
74eaa2a3e2 | ||
|
|
747bf5e02b | ||
|
|
8c18d0d40e | ||
|
|
bad4a24426 | ||
|
|
3e523e9725 | ||
|
|
b107aff150 | ||
|
|
1109a33edb | ||
|
|
e67aebbd3d | ||
|
|
2d30bd2155 | ||
|
|
f6f021f2a3 |
@@ -5,6 +5,11 @@ A Model Context Protocol (MCP) server designed to provide LLMs with efficient, t
|
||||
## ⚠️ AI SLOP DISCLAIMER
|
||||
This entire project was vibe-coded by an AI. It is 100% slop code. Use it at your own risk.
|
||||
|
||||
## 🚨 SECURITY WARNING
|
||||
**This server is designed to be run on `localhost` ONLY.**
|
||||
It contains tools (such as `read_image` and `list_directory`) that allow the LLM to read arbitrary files from your filesystem. If you expose this server to the network or a public IP, any user or compromised AI could potentially read sensitive system files (e.g., SSH keys, `/etc/passwd`).
|
||||
**NEVER run this server on a public-facing IP without implementing strict path validation.**
|
||||
|
||||
## Tools Overview
|
||||
|
||||
### 📁 `list_directory`
|
||||
@@ -64,9 +69,10 @@ This server requires Python 3.10+ and the following packages:
|
||||
* `starlette`: Lightweight ASGI framework.
|
||||
* `Pillow`: Image processing and thumbnail generation.
|
||||
* `requests`: For communicating with the Stable Diffusion API.
|
||||
* `httpx`: For asynchronous API requests (e.g., Wikipedia).
|
||||
|
||||
```bash
|
||||
pip install uvicorn starlette Pillow requests
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
### Setup
|
||||
|
||||
@@ -10,16 +10,19 @@ ROOT_DIR = Path(__file__).parent.resolve()
|
||||
HOST = "127.0.0.1"
|
||||
PORT = 8000
|
||||
request_host = ContextVar("request_host", default=f"{HOST}:{PORT}")
|
||||
LOG_LEVEL = "WARNING" # Options: "DEBUG", "INFO", "WARNING", "ERROR"
|
||||
LOG_LEVEL = "DEBUG" # Options: "DEBUG", "INFO", "WARNING", "ERROR"
|
||||
LOG_FILE = str(ROOT_DIR / "debug.log")
|
||||
USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64; rv:151.0) Gecko/20100101 Firefox/151.0" # Stealth User-Agent to bypass Wikipedia's bot detection
|
||||
USER_AGENT = "MooseCP/1.0 Local MCP Server (https://long-cat.net/)"
|
||||
|
||||
# --- Stable Diffusion Config ---
|
||||
SD_URL = "http://127.0.0.1:7860"
|
||||
MODEL_PRESETS_PATH = str(ROOT_DIR / "model_presets.toml")
|
||||
RES_PRESETS_PATH = str(ROOT_DIR / "resolution_presets.toml")
|
||||
TAG_DATABASE_PATH = "/home/matt/stable-diffusion-webui/extensions/a1111-sd-webui-tagcomplete/tags/danbooru.csv"
|
||||
TAG_DATABASE_PATH = str(ROOT_DIR / "danbooru.csv")
|
||||
TAG_SEARCH_LIMIT = 20
|
||||
ENABLE_TAG_WIKI = False
|
||||
DANBOORU_LOGIN = "" # Your Danbooru username (Optional)
|
||||
DANBOORU_API_KEY = "" # Your Danbooru API key (Optional)
|
||||
|
||||
# --- Model Specific Token Tuning (Tuned for Gemma 4) ---
|
||||
# Patch size is typically (clip.vision.patch_size * n_merge)
|
||||
|
||||
+140782
File diff suppressed because it is too large
Load Diff
+8
-4
@@ -10,9 +10,9 @@ This model is not aesthetic tuned, it must be given explicit tags for everything
|
||||
Has very excellent understanding of characters and artists down to extremely niche. Unless prompting an original character, their name is enough to decribe their appearance completely except for clothing.
|
||||
Accepts a list of comma-separated booru-style tags. Use spaces, not underscores, for tags. **Use the `search_tags` tool to verify your tags**.
|
||||
Prompts MUST follow this format: <1girl/1boy/1other/solo/couple/(can use multiple)>, <character(s)>, <series>, <artist>, <tags>
|
||||
Every prompt MUST include every one of the above sections.
|
||||
Every prompt MUST include every one of the above sections (the angle brackets are not part of the prompt).
|
||||
Quality tags such as "masterpiece", "best quality", "very awa", are a LAST RESORT, they override the artist tags. If absolutely required they should be prepended.
|
||||
It understands 'implicit' artist tags such as "official art" or "game cg" for the 'artist' immediately following the series.
|
||||
It understands every artist, so pick one appropriate for the image. If desired, it also knows 'implicit' artist tags such as "official art" or "game cg" for the 'artist' immediately following the series name.
|
||||
Natural language understanding is very limited, but can do things like "dark blue skirt" or natural language order of tags such as "lying, on bed".
|
||||
Do not use a negative prompt unless explicitly required to exclude something, your first prompt should have a blank negative prompt.
|
||||
Has the SDXL problem with hands, works best if hand posture is explicitly prompted.
|
||||
@@ -35,10 +35,14 @@ preset = "xl"
|
||||
"Rescale CFG" = 0.3
|
||||
|
||||
["Z-Image-Turbo"]
|
||||
description = "General image generation model. Aesthetic tuned, gets good results first try. NSFW is quite limited."
|
||||
description = "General image generation model. Aesthetic tuned, gets good results first try. Can do softcore NSFW, e.g. underwear, breasts, asses. No full-frontal nudity."
|
||||
guide = """
|
||||
This model is aesthetic tuned, regenerating with the same prompt will yield essentially the same image. Change the prompt before resubmitting.
|
||||
Understands natural language very well. Characters can be described by naming them and using this name later in the prompt. Longer, detailed prompts work better.
|
||||
It's a CFG 1.0 turbo model, the negative prompt has no effect.
|
||||
Understands natural language very well, uses Qwen 3 4B as the text encoder. The longer and more detailed prompt the better, take advantage of line breaks and formatting.
|
||||
If any part of the image is left out of the prompt, it will default to generic AI slop which is most NOT what you want. Be very explicit about each character's details. Appearance, ethnicity, age, individual outfit components, facial expression, pose, action, where they are looking, position in the image, etc. should ALL be included in the prompt. Characters can be referenced by naming them and using this name later in the prompt. This also helps the model avoid mixing traits between them.
|
||||
This also applies to the image itself. Composition, framing, lighting, image style, setting, background, etc. should all be explicitly specified in the prompt.
|
||||
Has some idiosyncrasies so you may need to iterate the prompt a few times to get around some weird artifacts. Think things like "green eyes" making them glow green, or "blush" making the entire face glow. Also really wants to make shirts tucked in for some reason. Examine the generated image closely and edit the prompt if needed.
|
||||
"""
|
||||
preset = "zit"
|
||||
filename = "z_image_turbo_bf16.safetensors"
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
uvicorn
|
||||
starlette
|
||||
requests
|
||||
httpx
|
||||
Pillow
|
||||
+3
-3
@@ -22,7 +22,7 @@ from .contact_sheet import handle as contact_sheet_handler
|
||||
from .list_directory import handle as list_directory_details_handler
|
||||
from .preview_image import handle as preview_image_handler
|
||||
from .get_text_context import handle as get_text_context_handler
|
||||
from .wikipedia import handle as wikipedia_handler
|
||||
from .browse_wikipedia import handle as wikipedia_handler
|
||||
from .get_model_info import handle as get_model_info_handler
|
||||
from .generate_image import handle as generate_image_handler
|
||||
from .search_tags import handle as search_tags_handler
|
||||
@@ -148,7 +148,7 @@ TOOL_REGISTRY = [
|
||||
"model_name": {"type": "string", "description": "The name of the model to use. Required."},
|
||||
"prompt": {"type": "string", "description": "The prompt for the image. Required."},
|
||||
"negative_prompt": {"type": "string", "description": "The negative prompt to exclude unwanted elements."},
|
||||
"resolution_preset": {"type": "string", "description": "A named resolution preset (e.g., 'square', 'portrait'). Available options depend on the model."},
|
||||
"resolution_preset": {"type": "string", "description": "A named resolution preset from the model info"},
|
||||
"cfg_scale": {"type": "number", "description": "CFG scale for prompt adherence. Usually should be left omitted to select the default."},
|
||||
},
|
||||
"required": ["model_name", "prompt", "resolution_preset"],
|
||||
@@ -157,7 +157,7 @@ TOOL_REGISTRY = [
|
||||
),
|
||||
Tool(
|
||||
name="search_tags",
|
||||
description="Searches the Danbooru tag database for tags matching a query. Returns the most popular tags including alias matches. Useful for finding the correct booru-style tags for anime models.",
|
||||
description="Searches the Danbooru tag database for tags matching a query. Returns the Danbooru wiki page on an exact match. On any query, returns a list of similar tags as well as aliases. Useful for finding the correct booru-style tags for anime models or getting more information.",
|
||||
schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import urllib.request
|
||||
import urllib.parse
|
||||
import httpx
|
||||
import json
|
||||
import re
|
||||
import asyncio
|
||||
@@ -13,22 +12,33 @@ def strip_html(text: str) -> str:
|
||||
"""Removes HTML tags from a string using regex to provide clean text to the LLM."""
|
||||
return re.sub(r'<[^>]*>', '', text)
|
||||
|
||||
def _make_request(params: Dict[str, Any]) -> Dict[str, Any]:
|
||||
def clean_wikitext(text: str) -> str:
|
||||
"""
|
||||
Synchronous helper to make the API request using urllib.
|
||||
urllib is used instead of httpx to avoid TLS/HTTP fingerprinting
|
||||
that triggers 403 Forbidden responses from Wikipedia.
|
||||
Removes the most distracting elements of raw Wikitext:
|
||||
1. HTML comments (<!-- ... -->)
|
||||
2. Citations (<ref /> and <ref>...</ref>)
|
||||
"""
|
||||
# Remove HTML comments
|
||||
text = re.sub(r'<!--.*?-->', '', text, flags=re.DOTALL)
|
||||
# Remove self-closing citations FIRST to prevent them being seen as opening tags
|
||||
text = re.sub(r'<ref[^>]*/>', '', text)
|
||||
# Remove paired citations
|
||||
text = re.sub(r'<ref[^>]*>.*?</ref>', '', text, flags=re.DOTALL)
|
||||
return text
|
||||
|
||||
async def _make_request(params: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Asynchronous helper to make the API request using httpx.
|
||||
A custom User-Agent is used to avoid bot detection.
|
||||
"""
|
||||
query_string = urllib.parse.urlencode(params)
|
||||
url = f"{API_URL}?{query_string}"
|
||||
|
||||
headers = {
|
||||
"User-Agent": config.USER_AGENT
|
||||
}
|
||||
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
with urllib.request.urlopen(req) as response:
|
||||
return json.loads(response.read().decode('utf-8'))
|
||||
async with httpx.AsyncClient(headers=headers, timeout=15.0) as client:
|
||||
response = await client.get(API_URL, params=params)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
async def _search(title: str, limit: int = 5, fallback: bool = False) -> str:
|
||||
"""
|
||||
@@ -42,8 +52,8 @@ async def _search(title: str, limit: int = 5, fallback: bool = False) -> str:
|
||||
"format": "json",
|
||||
"srlimit": limit
|
||||
}
|
||||
# Run synchronous urllib call in a thread to avoid blocking the event loop
|
||||
data = await asyncio.to_thread(_make_request, params)
|
||||
# Directly await the async request
|
||||
data = await _make_request(params)
|
||||
|
||||
search_results = data.get("query", {}).get("search", [])
|
||||
if not search_results:
|
||||
@@ -66,10 +76,11 @@ async def _search(title: str, limit: int = 5, fallback: bool = False) -> str:
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
async def _fetch_content(title: str, section_index: int) -> Optional[str]:
|
||||
async def _fetch_content(title: str, section_index: int) -> tuple[Optional[str], Optional[str]]:
|
||||
"""
|
||||
Retrieves raw Wikitext for a specific section.
|
||||
section_index=0 returns the lead section.
|
||||
Returns a tuple of (content, final_title).
|
||||
"""
|
||||
params = {
|
||||
"action": "query",
|
||||
@@ -80,27 +91,30 @@ async def _fetch_content(title: str, section_index: int) -> Optional[str]:
|
||||
"redirects": 1,
|
||||
"format": "json"
|
||||
}
|
||||
data = await asyncio.to_thread(_make_request, params)
|
||||
data = await _make_request(params)
|
||||
|
||||
pages = data.get("query", {}).get("pages", {})
|
||||
if not pages:
|
||||
return None
|
||||
return None, None
|
||||
|
||||
page_id = next(iter(pages))
|
||||
page = pages[page_id]
|
||||
final_title = page.get("title")
|
||||
|
||||
if "missing" in page:
|
||||
return None
|
||||
return None, None
|
||||
|
||||
revisions = page.get("revisions", [])
|
||||
if not revisions:
|
||||
return None
|
||||
return None, final_title
|
||||
|
||||
return revisions[0].get("*")
|
||||
content = revisions[0].get("*")
|
||||
return (clean_wikitext(content) if content else None), final_title
|
||||
|
||||
async def _fetch_toc(title: str) -> Optional[str]:
|
||||
async def _fetch_toc(title: str) -> tuple[Optional[str], Optional[str]]:
|
||||
"""
|
||||
Retrieves the Table of Contents data and formats it hierarchically.
|
||||
Returns a tuple of (toc, final_title).
|
||||
"""
|
||||
params = {
|
||||
"action": "parse",
|
||||
@@ -109,18 +123,19 @@ async def _fetch_toc(title: str) -> Optional[str]:
|
||||
"format": "json",
|
||||
"redirects": 1
|
||||
}
|
||||
data = await asyncio.to_thread(_make_request, params)
|
||||
data = await _make_request(params)
|
||||
|
||||
parse_data = data.get("parse")
|
||||
if not parse_data:
|
||||
return None
|
||||
return None, None
|
||||
|
||||
final_title = parse_data.get("title")
|
||||
toc_data = parse_data.get("tocdata", {})
|
||||
sections = toc_data.get("sections", [])
|
||||
if not sections:
|
||||
return "No table of contents found for this page."
|
||||
return "No table of contents found for this page.", final_title
|
||||
|
||||
lines = [f"Table of Contents for \"{parse_data.get('title', title)}\":"]
|
||||
lines = [f"Table of Contents for \"{final_title}\":"]
|
||||
for s in sections:
|
||||
level = s.get("tocLevel", 1)
|
||||
index = s.get("index")
|
||||
@@ -128,7 +143,7 @@ async def _fetch_toc(title: str) -> Optional[str]:
|
||||
indent = " " * (level - 1)
|
||||
lines.append(f"{indent}[{index}] {line}")
|
||||
|
||||
return "\n".join(lines)
|
||||
return "\n".join(lines), final_title
|
||||
|
||||
async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
@@ -143,18 +158,24 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
section_index = args.get("section_index")
|
||||
search_limit = args.get("search_limit", 5)
|
||||
|
||||
final_title = None
|
||||
|
||||
if mode == "search":
|
||||
result = await _search(title, search_limit, fallback=False)
|
||||
elif mode == "summary":
|
||||
# Lead section + ToC is the default 'summary' to guide the AI's next steps
|
||||
content = await _fetch_content(title, 0)
|
||||
content, title_from_content = await _fetch_content(title, 0)
|
||||
final_title = title_from_content
|
||||
if content is None:
|
||||
result = await _search(title, search_limit, fallback=True)
|
||||
else:
|
||||
toc = await _fetch_toc(title)
|
||||
toc, title_from_toc = await _fetch_toc(title)
|
||||
if title_from_toc:
|
||||
final_title = title_from_toc
|
||||
result = f"{content}\n\n---\n\n{toc}"
|
||||
elif mode == "toc":
|
||||
toc = await _fetch_toc(title)
|
||||
toc, title_from_toc = await _fetch_toc(title)
|
||||
final_title = title_from_toc
|
||||
if toc is None:
|
||||
result = await _search(title, search_limit, fallback=True)
|
||||
else:
|
||||
@@ -162,12 +183,16 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
elif mode == "section":
|
||||
if section_index is None:
|
||||
raise ToolError("Missing required parameter 'section_index' for mode='section'")
|
||||
content = await _fetch_content(title, int(section_index))
|
||||
content, title_from_content = await _fetch_content(title, int(section_index))
|
||||
final_title = title_from_content
|
||||
if content is None:
|
||||
result = await _search(title, search_limit, fallback=True)
|
||||
else:
|
||||
result = content
|
||||
else:
|
||||
raise ToolError(f"Invalid mode '{mode}'. Supported modes: summary, toc, section, search")
|
||||
|
||||
if final_title and final_title != title:
|
||||
result = f"Redirected to \"{final_title}\"\n\n{result}"
|
||||
|
||||
return [{"type": "text", "text": result}]
|
||||
+42
-11
@@ -32,7 +32,7 @@ async def ensure_model_state(model_name: str, model_preset: Dict[str, Any]):
|
||||
preset = model_preset.get("preset", "xl")
|
||||
await asyncio.to_thread(
|
||||
requests.post,
|
||||
f"{config.SD_URL}/api/predict/checkpoint_change",
|
||||
f"{config.SD_URL}/api/checkpoint_change",
|
||||
json={"data": [target_ckpt, preset]},
|
||||
timeout=30
|
||||
)
|
||||
@@ -45,7 +45,7 @@ async def ensure_model_state(model_name: str, model_preset: Dict[str, Any]):
|
||||
preset = model_preset.get("preset", "xl")
|
||||
await asyncio.to_thread(
|
||||
requests.post,
|
||||
f"{config.SD_URL}/api/predict/modules_change",
|
||||
f"{config.SD_URL}/api/modules_change",
|
||||
json={"data": [target_modules, preset]},
|
||||
timeout=30
|
||||
)
|
||||
@@ -81,16 +81,47 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
except Exception as e:
|
||||
raise ToolError(f"Failed to connect to Stable Diffusion server: {str(e)}")
|
||||
|
||||
# Build the label-to-index map
|
||||
# Build the label-to-index map and a sparse payload
|
||||
label_map = {}
|
||||
label_counts = {}
|
||||
|
||||
payload = [p["parameter_default"] for p in params_info]
|
||||
# Find the maximum param index to determine payload size
|
||||
max_param_idx = 0
|
||||
for p in params_info:
|
||||
name = p.get("parameter_name", "")
|
||||
if name.startswith("param_"):
|
||||
try:
|
||||
idx = int(name.replace("param_", ""))
|
||||
max_param_idx = max(max_param_idx, idx)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Initialize payload with Nones (size is max_idx + 1)
|
||||
payload = [None] * (max_param_idx + 1)
|
||||
|
||||
for idx, p in enumerate(params_info):
|
||||
label = p["label"]
|
||||
for idx_in_list, p in enumerate(params_info):
|
||||
name = p.get("parameter_name", "")
|
||||
label = p.get("label")
|
||||
default = p.get("parameter_default")
|
||||
|
||||
# Determine the absolute index in the payload
|
||||
if name == "id_task":
|
||||
abs_idx = 0
|
||||
elif name.startswith("param_"):
|
||||
try:
|
||||
abs_idx = int(name.replace("param_", ""))
|
||||
except ValueError:
|
||||
continue
|
||||
else:
|
||||
# Fallback for unexpected names, though unlikely
|
||||
continue
|
||||
|
||||
# Set the default value at the absolute index
|
||||
payload[abs_idx] = default
|
||||
|
||||
# Handle label mapping for AI overrides
|
||||
if not label or label.startswith("parameter_"):
|
||||
label = p["parameter_name"]
|
||||
label = name
|
||||
|
||||
if label in label_counts:
|
||||
label_counts[label] += 1
|
||||
@@ -99,7 +130,7 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
label_counts[label] = 0
|
||||
mapped_label = label
|
||||
|
||||
label_map[mapped_label] = idx
|
||||
label_map[mapped_label] = abs_idx
|
||||
|
||||
# 3. LOAD CONFIGS
|
||||
models_cfg = load_toml(config.MODEL_PRESETS_PATH)
|
||||
@@ -143,7 +174,7 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
except (ValueError, AttributeError):
|
||||
raise ToolError(f"Invalid resolution format for preset '{res_preset_name}': {res_val_str}. Expected 'WidthxHeight'.")
|
||||
else:
|
||||
raise ToolError(f"Resolution preset '{res_preset_name}' not found for this model. Available: {', '.join(res_set.keys())}")
|
||||
raise ToolError(f"Resolution preset '{res_preset_name}' not found for this model. You MUST call get_model_info with this model_name to see the available resolution presets.")
|
||||
else:
|
||||
raise ToolError(f"No resolution set configured for model '{model_name}'.")
|
||||
|
||||
@@ -163,8 +194,8 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
if "Seed" in label_map:
|
||||
payload[label_map["Seed"]] = -1
|
||||
|
||||
for _ in range(7):
|
||||
payload.insert(39, None)
|
||||
# The "Magic Number" splice is no longer needed as gaps are
|
||||
# automatically filled by the sparse-to-dense mapping.
|
||||
|
||||
# 6. EXECUTE GENERATION
|
||||
try:
|
||||
|
||||
@@ -24,7 +24,7 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
{
|
||||
"type": "text",
|
||||
"text": "Available models:\n\n" + "\n".join([f"- {m['name']}: {m['description']}" for m in catalog]) +
|
||||
"\n\nTo get a detailed prompting guide and available resolutions for a specific model, call this tool again with the 'model_name' argument."
|
||||
"\n\nBefore generating an image, you MUST call this tool again with your selected model as the 'model_name' argument."
|
||||
}
|
||||
]
|
||||
|
||||
@@ -41,7 +41,7 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
"type": "text",
|
||||
"text": f"Model '{model_name}' not found.\n\nAvailable models:\n\n" +
|
||||
"\n".join([f"- {m['name']}: {m['description']}" for m in catalog]) +
|
||||
"\n\nTo get a detailed prompting guide and available resolutions for a specific model, call this tool again with the 'model_name' argument."
|
||||
"\n\nBefore generating an image, you MUST call this tool again with selected model as the 'model_name' argument."
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
+110
-40
@@ -2,6 +2,8 @@ import csv
|
||||
import os
|
||||
import config
|
||||
import difflib
|
||||
import httpx
|
||||
import re
|
||||
from typing import List, Dict, Any, Optional
|
||||
from tools.utils import ToolError, format_count, get_type_suffix
|
||||
|
||||
@@ -41,14 +43,47 @@ def _load_tags() -> List[Dict[str, Any]]:
|
||||
|
||||
return tags
|
||||
|
||||
return tags
|
||||
|
||||
async def _fetch_wiki_info(tag_name: str) -> Optional[str]:
|
||||
"""Fetches the wiki description for a tag from Danbooru."""
|
||||
if not config.ENABLE_TAG_WIKI:
|
||||
return None
|
||||
|
||||
url = f"https://danbooru.donmai.us/wiki_pages/{tag_name}.json"
|
||||
try:
|
||||
# Use Basic Auth if credentials are provided
|
||||
auth = None
|
||||
if config.DANBOORU_LOGIN and config.DANBOORU_API_KEY:
|
||||
auth = (config.DANBOORU_LOGIN, config.DANBOORU_API_KEY)
|
||||
|
||||
async with httpx.AsyncClient(timeout=5.0) as client:
|
||||
# Using a browser-like UA to avoid potential blocks
|
||||
headers = {"User-Agent": config.USER_AGENT}
|
||||
resp = await client.get(url, headers=headers, auth=auth)
|
||||
|
||||
if resp.status_code == 403:
|
||||
return " (Wiki access blocked by Danbooru/Cloudflare)"
|
||||
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
wiki_body = data.get("wiki_page", {}).get("body")
|
||||
if wiki_body:
|
||||
# Strip HTML tags for the LLM
|
||||
return re.sub(r'<[^>]*>', '', wiki_body).strip()
|
||||
except Exception as e:
|
||||
return f" (Error fetching wiki: {str(e)})"
|
||||
return None
|
||||
|
||||
async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Searches the Danbooru tag database for tags matching a query.
|
||||
Returns the most popular tags including alias matches.
|
||||
Returns a unified list prioritized by substring matches then similarity.
|
||||
"""
|
||||
global _TAG_CACHE
|
||||
|
||||
query = args.get("query", "").lower()
|
||||
# Normalize query: treat spaces and underscores as identical
|
||||
query = args.get("query", "").lower().replace(" ", "_")
|
||||
if not query:
|
||||
raise ToolError("The 'query' argument is required.")
|
||||
|
||||
@@ -56,55 +91,90 @@ async def handle(args: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
if _TAG_CACHE is None:
|
||||
_TAG_CACHE = _load_tags()
|
||||
|
||||
matches = []
|
||||
# 1. Find Substring/Alias Matches (High Priority)
|
||||
substring_matches = []
|
||||
for tag in _TAG_CACHE:
|
||||
# Check for match in name or aliases
|
||||
is_direct = query in tag["name_lower"]
|
||||
is_alias = any(query in alias.strip() for alias in tag["aliases"])
|
||||
|
||||
|
||||
if is_direct or is_alias:
|
||||
matches.append({
|
||||
"name": tag["name"],
|
||||
"type": tag["type"],
|
||||
"count": tag["count"],
|
||||
# Find which alias actually matched for reporting
|
||||
matched_alias = ""
|
||||
if not is_direct:
|
||||
matched_alias = next((a.strip() for a in tag["aliases"] if query in a.strip()), query)
|
||||
|
||||
substring_matches.append({
|
||||
"tag": tag,
|
||||
"matched_via": "name" if is_direct else "alias",
|
||||
"alias_match": "" if is_direct else next((a.strip() for a in tag["aliases"] if query in a.strip()), query)
|
||||
"alias_match": matched_alias
|
||||
})
|
||||
|
||||
# Sort substring matches by count descending
|
||||
substring_matches.sort(key=lambda x: x["tag"]["count"], reverse=True)
|
||||
|
||||
# Sort by count descending
|
||||
matches.sort(key=lambda x: x["count"], reverse=True)
|
||||
# 2. Find Similarity Matches (Low Priority)
|
||||
all_names_lower = [t["name_lower"] for t in _TAG_CACHE]
|
||||
similar_names_lower = difflib.get_close_matches(query, all_names_lower, n=config.TAG_SEARCH_LIMIT, cutoff=0.5)
|
||||
similar_matches = []
|
||||
for s_lower in similar_names_lower:
|
||||
tag = next((t for t in _TAG_CACHE if t["name_lower"] == s_lower), None)
|
||||
if tag:
|
||||
similar_matches.append(tag)
|
||||
|
||||
# Format top results
|
||||
results = []
|
||||
for m in matches[:config.TAG_SEARCH_LIMIT]:
|
||||
count_fmt = format_count(str(m["count"]))
|
||||
type_sfx = get_type_suffix(m["type"])
|
||||
# Build Unified List
|
||||
final_results = []
|
||||
|
||||
# Add substring matches first
|
||||
for m in substring_matches:
|
||||
final_results.append(m)
|
||||
if len(final_results) >= config.TAG_SEARCH_LIMIT:
|
||||
break
|
||||
|
||||
# Fill remaining slots with similar matches
|
||||
if len(final_results) < config.TAG_SEARCH_LIMIT:
|
||||
mentioned_names = {m["tag"]["name_lower"] for m in final_results}
|
||||
for tag in similar_matches:
|
||||
if tag["name_lower"] not in mentioned_names:
|
||||
final_results.append({"tag": tag, "matched_via": "similarity", "alias_match": ""})
|
||||
if len(final_results) >= config.TAG_SEARCH_LIMIT:
|
||||
break
|
||||
|
||||
if not final_results:
|
||||
return [{"type": "text", "text": f"No tags found matching '{query}'."}]
|
||||
|
||||
# Format output lines
|
||||
output_lines = []
|
||||
|
||||
# Special Case: Exact Match Wiki Header
|
||||
# Check if the very first result is an exact match
|
||||
first_res = final_results[0]
|
||||
if first_res["tag"]["name_lower"] == query:
|
||||
exact_tag = first_res["tag"]
|
||||
count_fmt = format_count(str(exact_tag["count"]))
|
||||
type_sfx = get_type_suffix(exact_tag["type"])
|
||||
output_lines.append(f"Exact Match: {exact_tag['name']} ({count_fmt}){type_sfx}")
|
||||
|
||||
if m["matched_via"] == "alias":
|
||||
line = f"{m['alias_match']} → {m['name']} ({count_fmt}){type_sfx}"
|
||||
wiki_info = await _fetch_wiki_info(exact_tag["name"])
|
||||
if wiki_info:
|
||||
output_lines.append(f"Wiki: {wiki_info}\n")
|
||||
else:
|
||||
line = f"{m['name']} ({count_fmt}){type_sfx}"
|
||||
output_lines.append("") # spacer
|
||||
|
||||
# List the tags
|
||||
for res in final_results:
|
||||
tag = res["tag"]
|
||||
count_fmt = format_count(str(tag["count"]))
|
||||
type_sfx = get_type_suffix(tag["type"])
|
||||
|
||||
results.append(line)
|
||||
# If this is the exact match we already listed in the header, skip it
|
||||
if first_res["tag"]["name_lower"] == query and tag["name_lower"] == query:
|
||||
continue
|
||||
|
||||
if not results:
|
||||
# Attempt to find similar tags using difflib
|
||||
all_names_lower = [t["name_lower"] for t in _TAG_CACHE]
|
||||
suggestions_lower = difflib.get_close_matches(query, all_names_lower, n=config.TAG_SEARCH_LIMIT, cutoff=0.5)
|
||||
if res["matched_via"] == "alias":
|
||||
line = f"- {res['alias_match']} → {tag['name']} ({count_fmt}){type_sfx}"
|
||||
else:
|
||||
line = f"- {tag['name']} ({count_fmt}){type_sfx}"
|
||||
|
||||
if not suggestions_lower:
|
||||
return [{"type": "text", "text": f"No tags found matching '{query}'."}]
|
||||
|
||||
# Map lowercased suggestions back to original tag objects
|
||||
suggestions = []
|
||||
for s_lower in suggestions_lower:
|
||||
# Find the first tag that matches this lowercased name
|
||||
tag = next((t for t in _TAG_CACHE if t["name_lower"] == s_lower), None)
|
||||
if tag:
|
||||
count_fmt = format_count(str(tag["count"]))
|
||||
type_sfx = get_type_suffix(tag["type"])
|
||||
suggestions.append(f"{tag['name']} ({count_fmt}){type_sfx}")
|
||||
|
||||
return [{"type": "text", "text": f"No exact matches for '{query}'. Did you mean:\n" + "\n".join(suggestions)}]
|
||||
output_lines.append(line)
|
||||
|
||||
return [{"type": "text", "text": "Top matches:\n" + "\n".join(results)}]
|
||||
return [{"type": "text", "text": "\n".join(output_lines)}]
|
||||
|
||||
Reference in New Issue
Block a user