mirror of
https://github.com/vladmandic/automatic
synced 2026-09-20 01:31:13 +02:00
feat(api): add tag dictionary API and settings for prompt autocomplete
This commit is contained in:
+7
-1
@@ -1,9 +1,10 @@
|
||||
import os
|
||||
from threading import Lock
|
||||
from secrets import compare_digest
|
||||
from fastapi import FastAPI, APIRouter, Depends, Request
|
||||
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
||||
from fastapi.exceptions import HTTPException
|
||||
from modules import errors, shared
|
||||
from modules import errors, shared, paths
|
||||
from modules.logger import log
|
||||
from modules.api import models, endpoints, script, helpers, server, generate, process, control, docs, gpu
|
||||
|
||||
@@ -117,6 +118,11 @@ class Api:
|
||||
from modules.api import loras
|
||||
loras.register_api(self.app)
|
||||
|
||||
# dicts api
|
||||
from modules.api import dicts as dicts_api
|
||||
dicts_api.init(getattr(shared.opts, 'dicts_dir', '') or os.path.join(paths.models_path, 'dicts'))
|
||||
dicts_api.register_api(self.app)
|
||||
|
||||
# gallery api
|
||||
from modules.api import gallery
|
||||
gallery.register_api(self.app)
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
"""V1 dictionary / tag autocomplete endpoints.
|
||||
|
||||
Serves pre-built tag dictionaries (Danbooru, e621, natural language, artists)
|
||||
from JSON files in the configured dicts directory.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
from fastapi.exceptions import HTTPException
|
||||
|
||||
from modules.api.models import ItemDict, ItemDictContent
|
||||
|
||||
dicts_dir: str = ""
|
||||
cache: dict[str, dict] = {}
|
||||
|
||||
|
||||
def init(path: str) -> None:
|
||||
"""Set the dicts directory path. Called once during API registration."""
|
||||
global dicts_dir # noqa: PLW0603
|
||||
dicts_dir = path
|
||||
|
||||
|
||||
def get_cached(name: str) -> dict:
|
||||
"""Load a dict file, returning cached version if file hasn't changed."""
|
||||
if '/' in name or '\\' in name or '..' in name:
|
||||
raise HTTPException(status_code=400, detail="Invalid dict name")
|
||||
path = os.path.join(dicts_dir, f"{name}.json")
|
||||
if not os.path.isfile(path):
|
||||
cache.pop(name, None)
|
||||
raise HTTPException(status_code=404, detail=f"Dict not found: {name}")
|
||||
stat = os.stat(path)
|
||||
entry = cache.get(name)
|
||||
if entry and entry['mtime'] == stat.st_mtime:
|
||||
return entry
|
||||
with open(path, encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
entry = {
|
||||
'mtime': stat.st_mtime,
|
||||
'size': stat.st_size,
|
||||
'meta': {
|
||||
'name': data.get('name', name),
|
||||
'version': data.get('version', ''),
|
||||
'tag_count': len(data.get('tags', [])),
|
||||
'categories': {
|
||||
str(k): v.get('name', str(k)) if isinstance(v, dict) else str(v)
|
||||
for k, v in data.get('categories', {}).items()
|
||||
},
|
||||
},
|
||||
'content': data,
|
||||
}
|
||||
cache[name] = entry
|
||||
return entry
|
||||
|
||||
|
||||
def list_dicts_sync() -> list[ItemDict]:
|
||||
"""Scan dicts directory and return metadata for each dict file."""
|
||||
if not dicts_dir or not os.path.isdir(dicts_dir):
|
||||
return []
|
||||
items = []
|
||||
for filename in sorted(os.listdir(dicts_dir)):
|
||||
if not filename.endswith('.json') or filename.startswith('.'):
|
||||
continue
|
||||
name = filename.rsplit('.', 1)[0]
|
||||
try:
|
||||
entry = get_cached(name)
|
||||
meta = entry['meta']
|
||||
items.append(ItemDict(
|
||||
name=meta['name'],
|
||||
version=meta['version'],
|
||||
tag_count=meta['tag_count'],
|
||||
categories=meta['categories'],
|
||||
size=entry['size'],
|
||||
))
|
||||
except Exception:
|
||||
pass
|
||||
return items
|
||||
|
||||
|
||||
async def list_dicts() -> list[ItemDict]:
|
||||
"""List available tag dictionaries."""
|
||||
return await asyncio.to_thread(list_dicts_sync)
|
||||
|
||||
|
||||
async def get_dict(name: str) -> ItemDictContent:
|
||||
"""Get full dict content by name."""
|
||||
def _load():
|
||||
return get_cached(name)
|
||||
try:
|
||||
entry = await asyncio.to_thread(_load)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e)) from e
|
||||
content = entry['content']
|
||||
return ItemDictContent(
|
||||
name=content.get('name', name),
|
||||
version=content.get('version', ''),
|
||||
categories=content.get('categories', {}),
|
||||
tags=content.get('tags', []),
|
||||
)
|
||||
|
||||
|
||||
def register_api(app):
|
||||
app.add_api_route("/sdapi/v1/dicts", list_dicts, methods=["GET"], response_model=list[ItemDict], tags=["Enumerators"])
|
||||
app.add_api_route("/sdapi/v1/dicts/{name}", get_dict, methods=["GET"], response_model=ItemDictContent, tags=["Enumerators"])
|
||||
@@ -509,6 +509,19 @@ class ItemLoadedModel(BaseModel):
|
||||
dtype: Optional[str] = Field(default=None, title="Dtype", description="Effective data type (e.g., float16, nf4)")
|
||||
extra: Optional[dict] = Field(default=None, title="Extra metadata", description="Additional metadata (role, class, quantization method, etc.)")
|
||||
|
||||
class ItemDict(BaseModel):
|
||||
name: str = Field(title="Name", description="Dictionary identifier (filename without extension)")
|
||||
version: str = Field(default="", title="Version", description="Dictionary format version string")
|
||||
tag_count: int = Field(default=0, title="Tag count", description="Number of tags in this dictionary")
|
||||
categories: dict = Field(default_factory=dict, title="Categories", description="Category ID to display name mapping")
|
||||
size: int = Field(default=0, title="Size", description="File size in bytes")
|
||||
|
||||
class ItemDictContent(BaseModel):
|
||||
name: str = Field(title="Name", description="Dictionary identifier")
|
||||
version: str = Field(default="", title="Version", description="Dictionary format version string")
|
||||
categories: dict = Field(default_factory=dict, title="Categories", description="Category definitions with name and color")
|
||||
tags: list = Field(default_factory=list, title="Tags", description="Tag entries as [name, category_id, post_count] tuples")
|
||||
|
||||
# helper function
|
||||
|
||||
def create_model_from_signature(func: Callable, model_name: str, base_model: type[BaseModel] = BaseModel, additional_fields: list | None = None, exclude_fields: list[str] | None = None) -> type[BaseModel]:
|
||||
|
||||
Reference in New Issue
Block a user