Vito
2b8126b7be
* feat(images): card selection checkbox and the concept selection bar
Every Images card's overlay starts with the concept's "Select <filename>"
checkbox before the star. Checked image keys form the page's selection,
kept across Load more, infinite scroll and bounded re-mounting and cleared
whenever the collection is replaced. The #bulkbar between the filter row
and the gallery shows "N selected" with Set category, Export metadata and
Clear (the first two are wired by the next tasks); Clear empties it.
The interim D8 star rule is removed, and the concept comparison now
measures the overlay with its checkbox and the bar with one card checked.
* feat(images): export the selected images' metadata as one JSON bundle
The selection bar's Export metadata reads each checked image's inspection,
at most four at a time, and downloads prompt-manager-images-metadata.json:
format, version 1, exported_at and, in listing order, the same envelope the
inspector's "Download metadata JSON" saves (one shared builder,
metadataEnvelope). The button is aria-disabled and ignores a second press
while it runs, the selection is kept, and the toast reports the exported
count, the images that could not be read, or that none could (no download).
Also: the bar and its count are built only on the Images page, the
Library-embedded gallery is asserted to have no checkbox or bar, a sort
change is asserted to clear the selection, the checkbox label test asserts
the toggle and an unchanged selection and inspector, and TESTING.md records
that a return to Images keeps the selection while a reload starts empty.
* feat(images): set the category of the selected images' linked prompts
The selection bar's Set category opens the concept's simpleInput dialog
("Categorize selected images", one Category field, Cancel / Apply) and
posts the selection's distinct linked positive ids to
/prompt_manager/bulk/category. Images have no category of their own; the
listing's category filter matches the linked positive's, so that is what
changes. With no linked image checked nothing is posted and a toast says
so; on success the toast counts the prompts and any skipped images, then
the categories and the listing reload (clearing the selection); a failure
keeps the selection. The button ignores a second press while it runs.
The bulk category route now rejects anything but a non-empty list of at
most 1000 positive integer ids (deduplicated in order) and validates the
category with validate_category; bad JSON or a non-object body is a 400.
The response shape and the legacy admin request are unchanged.
* test(images): zoomed card keyboard order follows the concept's checkbox before the star
The card's Tab order is image, Select checkbox, star, Inspect (concept app.js:398), both ways; List hides the overlays as the concept does, so there the image is followed by Inspect.
* fix(api): answer 400 for bulk category ids beyond SQLite's INTEGER range
An id of 2**63 or more passed the parser and made sqlite3 raise OverflowError, which the route turned into a 500.
* refactor(images): move the selection bar into images-selection.mjs and harden its actions
- images.mjs is back under 800 lines; the bar, its checked set, Export metadata and Set category live in images-selection.mjs.
- Busy flags release in finally; the dead guards are gone and the category targets are taken when the modal dialog opens.
- The category write no longer carries the page signal; leaving Images mid-write drops the saved visit so the return reloads.
- The categories and the listing reload together after Set category.
- The Category field stops at the server's 100 characters; the server's updated_count is a tested pure helper.
- New browser tests: leaving mid-export and mid-write, Blob URL revoke, updated_count 0 and a 400 rejection.
- Additive coverage baseline entries for images-category, images-export and images-selection.
- TESTING.md D8 records the dialog deviations and these rulings.
1260 lines
46 KiB
Python
1260 lines
46 KiB
Python
"""Prompt API routes for PromptManager."""
|
|
|
|
import datetime
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
|
|
from aiohttp import web
|
|
|
|
try:
|
|
from ...database.pairs import PairDependencyError
|
|
except ImportError:
|
|
from database.pairs import PairDependencyError
|
|
from urllib.parse import quote
|
|
|
|
# Upper bounds on page sizes and offsets for every list endpoint. The offset
|
|
# ceiling keeps values inside SQLite's 64-bit range and bounds skip work.
|
|
MAX_PAGE_LIMIT = 500
|
|
MAX_PAGE_OFFSET = 10_000_000
|
|
|
|
# The export walks the whole table in pages of this size instead of asking
|
|
# for one giant result set (and silently truncating large libraries).
|
|
EXPORT_PAGE_SIZE = 1000
|
|
|
|
# The most prompts one bulk category request may change.
|
|
BULK_CATEGORY_MAX_PROMPTS = 1000
|
|
|
|
# The largest value SQLite's INTEGER can bind; a larger id raises OverflowError.
|
|
MAX_SQLITE_INT = 2**63 - 1
|
|
|
|
|
|
def parse_bulk_prompt_ids(value, cap=BULK_CATEGORY_MAX_PROMPTS):
|
|
"""A bulk request's ``prompt_ids``: distinct positive ints, first-seen order.
|
|
|
|
Raises ``ValueError`` unless ``value`` is a non-empty list of at most
|
|
``cap`` distinct positive integers no larger than SQLite's INTEGER
|
|
(booleans are not integers here).
|
|
"""
|
|
if not isinstance(value, list) or not value:
|
|
raise ValueError("prompt_ids must be a non-empty list")
|
|
if any(type(pid) is not int or not 1 <= pid <= MAX_SQLITE_INT for pid in value):
|
|
raise ValueError("prompt_ids must be positive integers")
|
|
ids = list(dict.fromkeys(value))
|
|
if len(ids) > cap:
|
|
raise ValueError(f"At most {cap} prompts can be changed at once")
|
|
return ids
|
|
|
|
|
|
def parse_page_params(query, default_limit=50, max_limit=MAX_PAGE_LIMIT):
|
|
"""Return ``(limit, offset)`` from a query mapping, clamped to safe bounds.
|
|
|
|
``limit`` is clamped to ``[1, max_limit]`` and ``offset`` to
|
|
``[0, MAX_PAGE_OFFSET]``. Non-integer values raise ``ValueError`` so the
|
|
caller can answer 400.
|
|
"""
|
|
limit = int(query.get("limit", default_limit))
|
|
offset = int(query.get("offset", 0))
|
|
return max(1, min(limit, max_limit)), max(0, min(offset, MAX_PAGE_OFFSET))
|
|
|
|
|
|
def bad_request(message):
|
|
return web.json_response({"success": False, "error": message}, status=400)
|
|
|
|
|
|
def tags_error(tags):
|
|
"""A 400 response when *tags* fail ``validate_tags``, else None."""
|
|
try:
|
|
validate_tags(tags)
|
|
except ValueError as exc:
|
|
return bad_request(str(exc))
|
|
return None
|
|
|
|
|
|
def safe_error_message(exc):
|
|
"""Describe *exc* for a client without leaking absolute server paths.
|
|
|
|
OSError carries the offending path in ``filename``; only its basename
|
|
is echoed back together with ``strerror``.
|
|
"""
|
|
if isinstance(exc, OSError):
|
|
reason = exc.strerror or exc.__class__.__name__
|
|
if exc.filename:
|
|
return f"{reason}: {os.path.basename(str(exc.filename))}"
|
|
return reason
|
|
return str(exc) or exc.__class__.__name__
|
|
|
|
|
|
def thumbnail_url_for(thumb_rel, thumb_abs):
|
|
"""Serve URL for an existing thumbnail, or None when the file is missing.
|
|
|
|
The URL carries the file's mtime as ``?v=``: thumbnails are served with a
|
|
long cache lifetime, so a regenerated file must get a new URL or browsers
|
|
keep showing whatever they cached for the old one.
|
|
"""
|
|
try:
|
|
version = int(thumb_abs.stat().st_mtime)
|
|
except OSError:
|
|
return None
|
|
return f"/prompt_manager/images/serve/{quote(thumb_rel, safe='/')}?v={version}"
|
|
|
|
|
|
def publish_image_paths(images, public_path):
|
|
"""Image dicts for a response: ``image_path`` in its public (ComfyUI
|
|
relative) form and the absolute ``file_path`` duplicate removed.
|
|
|
|
``url``/``thumbnail_url`` added by enrichment are left untouched, so the
|
|
frontend keeps working without ever seeing the server's directory layout.
|
|
"""
|
|
published = []
|
|
for image in images:
|
|
if not isinstance(image, dict):
|
|
published.append(image)
|
|
continue
|
|
image = {k: v for k, v in image.items() if k != "file_path"}
|
|
if "image_path" in image:
|
|
image["image_path"] = public_path(image["image_path"])
|
|
published.append(image)
|
|
return published
|
|
|
|
|
|
try:
|
|
from ...utils.validators import (
|
|
validate_prompt_text,
|
|
validate_rating,
|
|
validate_tags,
|
|
validate_category,
|
|
sanitize_input,
|
|
)
|
|
from ...utils.hashing import generate_prompt_hash
|
|
except ImportError:
|
|
from utils.validators import (
|
|
validate_prompt_text,
|
|
validate_rating,
|
|
validate_tags,
|
|
validate_category,
|
|
sanitize_input,
|
|
)
|
|
from utils.hashing import generate_prompt_hash
|
|
|
|
|
|
class PromptRoutesMixin:
|
|
"""Mixin providing prompt-related API endpoints."""
|
|
|
|
def _register_prompt_routes(self, routes):
|
|
@routes.get("/prompt_manager/search")
|
|
async def search_prompts_route(request):
|
|
return await self.search_prompts(request)
|
|
|
|
@routes.get("/prompt_manager/recent")
|
|
async def get_recent_prompts_route(request):
|
|
return await self.get_recent_prompts(request)
|
|
|
|
@routes.get("/prompt_manager/categories")
|
|
async def get_categories_route(request):
|
|
return await self.get_categories(request)
|
|
|
|
# Tag management endpoints (must be registered BEFORE /prompt_manager/tags)
|
|
@routes.get("/prompt_manager/tags/stats")
|
|
async def get_tags_stats_route(request):
|
|
return await self.get_tags_stats(request)
|
|
|
|
@routes.get("/prompt_manager/tags/filter")
|
|
async def get_tags_filter_route(request):
|
|
return await self.get_tags_filter(request)
|
|
|
|
@routes.get("/prompt_manager/tags/suggest")
|
|
async def suggest_tags_route(request):
|
|
return await self.suggest_tags(request)
|
|
|
|
# Bulk tag operations (register BEFORE {tag_name} to avoid path param match)
|
|
@routes.post("/prompt_manager/tags/merge")
|
|
async def merge_tags_route(request):
|
|
return await self.merge_tags_endpoint(request)
|
|
|
|
@routes.put("/prompt_manager/tags/{tag_name}")
|
|
async def rename_tag_route(request):
|
|
return await self.rename_tag_endpoint(request)
|
|
|
|
@routes.delete("/prompt_manager/tags/{tag_name}")
|
|
async def delete_tag_route(request):
|
|
return await self.delete_tag_endpoint(request)
|
|
|
|
@routes.get("/prompt_manager/tags/{tag_name}/prompts")
|
|
async def get_tag_prompts_route(request):
|
|
return await self.get_tag_prompts(request)
|
|
|
|
@routes.get("/prompt_manager/tags")
|
|
async def get_tags_route(request):
|
|
return await self.get_tags(request)
|
|
|
|
@routes.post("/prompt_manager/save")
|
|
async def save_prompt_route(request):
|
|
return await self.save_prompt(request)
|
|
|
|
@routes.delete("/prompt_manager/delete/{prompt_id}")
|
|
async def delete_prompt_route(request):
|
|
return await self.delete_prompt(request)
|
|
|
|
# Individual prompt management
|
|
@routes.put("/prompt_manager/prompts/{prompt_id}")
|
|
async def update_prompt_route(request):
|
|
return await self.update_prompt(request)
|
|
|
|
@routes.put("/prompt_manager/prompts/{prompt_id}/rating")
|
|
async def update_rating_route(request):
|
|
return await self.update_prompt_rating(request)
|
|
|
|
@routes.post("/prompt_manager/prompts/{prompt_id}/tags")
|
|
async def add_tag_route(request):
|
|
return await self.add_prompt_tag(request)
|
|
|
|
@routes.delete("/prompt_manager/prompts/{prompt_id}/tags")
|
|
async def remove_tag_route(request):
|
|
return await self.remove_prompt_tag(request)
|
|
|
|
@routes.post("/prompt_manager/prompts/tags")
|
|
async def add_tags_to_prompt_route(request):
|
|
return await self.add_tags_to_prompt(request)
|
|
|
|
# Bulk operations
|
|
@routes.post("/prompt_manager/bulk/delete")
|
|
async def bulk_delete_route(request):
|
|
return await self.bulk_delete_prompts(request)
|
|
|
|
@routes.post("/prompt_manager/bulk/tags")
|
|
async def bulk_add_tags_route(request):
|
|
return await self.bulk_add_tags(request)
|
|
|
|
@routes.post("/prompt_manager/bulk/category")
|
|
async def bulk_set_category_route(request):
|
|
return await self.bulk_set_category(request)
|
|
|
|
@routes.get("/prompt_manager/subfolders")
|
|
async def get_subfolders_route(request):
|
|
return await self.get_subfolders(request)
|
|
|
|
# Export functionality
|
|
@routes.get("/prompt_manager/export")
|
|
async def export_prompts_route(request):
|
|
return await self.export_prompts(request)
|
|
|
|
def _present_prompts(self, prompts):
|
|
"""Prompts ready for a response: image urls added, server paths hidden."""
|
|
self._enrich_prompt_images(prompts)
|
|
for prompt in prompts:
|
|
prompt["images"] = publish_image_paths(
|
|
prompt.get("images", []), self._public_path
|
|
)
|
|
return prompts
|
|
|
|
async def search_prompts(self, request):
|
|
"""Search for prompts using multiple filter criteria."""
|
|
try:
|
|
text = request.query.get("text", "").strip()
|
|
category = request.query.get("category", "").strip()
|
|
tags_str = request.query.get("tags", "").strip()
|
|
min_rating = request.query.get("min_rating", 0)
|
|
try:
|
|
limit, offset = parse_page_params(request.query)
|
|
except ValueError:
|
|
return bad_request("limit and offset must be integers")
|
|
|
|
folder = request.query.get("folder", "").strip() or None
|
|
# Validated against a whitelist in the database layer
|
|
sort = request.query.get("sort") or None
|
|
|
|
tags = None
|
|
if tags_str:
|
|
tags = [tag.strip() for tag in tags_str.split(",") if tag.strip()]
|
|
|
|
try:
|
|
min_rating = int(min_rating) if min_rating else None
|
|
except ValueError:
|
|
min_rating = None
|
|
|
|
results = await self._run_in_executor(
|
|
self.db.search_prompts,
|
|
text=text if text else None,
|
|
category=category if category else None,
|
|
tags=tags,
|
|
rating_min=min_rating,
|
|
limit=limit,
|
|
offset=offset,
|
|
folder=folder,
|
|
sort=sort,
|
|
)
|
|
self._present_prompts(results)
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"results": results,
|
|
"count": len(results),
|
|
"pagination": {
|
|
"limit": limit,
|
|
"offset": offset,
|
|
"count": len(results),
|
|
},
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Search error: {e}", exc_info=True)
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Search failed: {safe_error_message(e)}",
|
|
"results": [],
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def get_subfolders(self, request):
|
|
"""Get distinct subfolder values derived from generated image paths."""
|
|
try:
|
|
from ..config import GalleryConfig
|
|
|
|
root_dirs = list(GalleryConfig.MONITORING_DIRECTORIES) or None
|
|
include_ancestors = (
|
|
request.query.get("include_ancestors", "").lower() == "true"
|
|
)
|
|
subfolders = await self._run_in_executor(
|
|
self.db.get_prompt_subfolders, root_dirs, include_ancestors
|
|
)
|
|
return web.json_response({"success": True, "subfolders": subfolders})
|
|
except Exception as e:
|
|
self.logger.error(f"Subfolders error: {e}")
|
|
return web.json_response(
|
|
{"success": False, "error": safe_error_message(e)}, status=500
|
|
)
|
|
|
|
async def get_recent_prompts(self, request):
|
|
"""Retrieve prompts with pagination and an optional sort (default newest
|
|
first)."""
|
|
try:
|
|
try:
|
|
limit, offset = parse_page_params(request.query)
|
|
page = int(request.query.get("page", 1))
|
|
except ValueError:
|
|
return bad_request("limit, offset and page must be integers")
|
|
|
|
if page > 1 and offset == 0:
|
|
offset = min((page - 1) * limit, MAX_PAGE_OFFSET)
|
|
|
|
# Validated against a whitelist in the database layer
|
|
sort = request.query.get("sort") or None
|
|
results = await self._run_in_executor(
|
|
self.db.get_recent_prompts, limit=limit, offset=offset, sort=sort
|
|
)
|
|
self._present_prompts(results["prompts"])
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"results": results["prompts"],
|
|
"pagination": {
|
|
"total": results["total"],
|
|
"limit": results["limit"],
|
|
"offset": results["offset"],
|
|
"page": results["page"],
|
|
"total_pages": results["total_pages"],
|
|
"has_more": results["has_more"],
|
|
"count": len(results["prompts"]),
|
|
},
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Recent prompts error: {e}", exc_info=True)
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to get recent prompts: {safe_error_message(e)}",
|
|
"results": [],
|
|
"pagination": {"total": 0, "page": 1, "total_pages": 0},
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def get_categories(self, request):
|
|
"""Retrieve all available prompt categories."""
|
|
try:
|
|
categories = await self._run_in_executor(self.db.get_all_categories)
|
|
return web.json_response({"success": True, "categories": categories})
|
|
except Exception as e:
|
|
self.logger.error(f"Categories error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to get categories: {safe_error_message(e)}",
|
|
"categories": [],
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def get_tags(self, request):
|
|
"""Retrieve all available prompt tags."""
|
|
try:
|
|
tags = await self._run_in_executor(self.db.get_all_tags)
|
|
return web.json_response({"success": True, "tags": tags})
|
|
except Exception as e:
|
|
self.logger.error(f"Tags error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to get tags: {safe_error_message(e)}",
|
|
"tags": [],
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def get_tags_stats(self, request):
|
|
"""Get tags with usage counts, search, sort, and pagination."""
|
|
try:
|
|
try:
|
|
limit, offset = parse_page_params(request.query)
|
|
except ValueError:
|
|
return bad_request("limit and offset must be integers")
|
|
search = request.query.get("search", "").strip() or None
|
|
sort = request.query.get("sort", "alpha_asc")
|
|
|
|
result = await self._run_in_executor(
|
|
self.db.get_tags_with_counts, limit, offset, search, sort
|
|
)
|
|
untagged_count = await self._run_in_executor(
|
|
self.db.get_untagged_prompts_count
|
|
)
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"tags": result["tags"],
|
|
"untagged_count": untagged_count,
|
|
"pagination": {
|
|
"total": result["total"],
|
|
"limit": result["limit"],
|
|
"offset": result["offset"],
|
|
"has_more": result["has_more"],
|
|
},
|
|
}
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Tags stats error: {e}", exc_info=True)
|
|
return web.json_response(
|
|
{"success": False, "error": safe_error_message(e)}, status=500
|
|
)
|
|
|
|
async def get_tag_prompts(self, request):
|
|
"""Get prompts for a single tag."""
|
|
try:
|
|
# aiohttp has already percent-decoded match_info; decoding again
|
|
# would turn a tag literally named "%41" into "A".
|
|
tag_name = request.match_info.get("tag_name", "")
|
|
if not tag_name:
|
|
return web.json_response(
|
|
{"success": False, "error": "Tag name required"}, status=400
|
|
)
|
|
|
|
try:
|
|
limit, offset = parse_page_params(request.query, default_limit=20)
|
|
except ValueError:
|
|
return bad_request("limit and offset must be integers")
|
|
|
|
result = await self._run_in_executor(
|
|
self.db.get_prompts_by_tags, [tag_name], "and", limit, offset
|
|
)
|
|
self._present_prompts(result["prompts"])
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"tag": tag_name,
|
|
"prompts": result["prompts"],
|
|
"pagination": {
|
|
"total": result["total"],
|
|
"limit": result["limit"],
|
|
"offset": result["offset"],
|
|
"has_more": result["has_more"],
|
|
},
|
|
}
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Tag prompts error: {e}", exc_info=True)
|
|
return web.json_response(
|
|
{"success": False, "error": safe_error_message(e)}, status=500
|
|
)
|
|
|
|
async def suggest_tags(self, request):
|
|
"""Autocomplete for the tag search box.
|
|
|
|
``q`` is the prefix being typed, ``with`` the comma-separated tags
|
|
already entered: only tags that co-occur with all of them are offered.
|
|
"""
|
|
try:
|
|
prefix = request.query.get("q", "").strip()[:100]
|
|
context = []
|
|
seen = set()
|
|
for raw in request.query.get("with", "").split(","):
|
|
name = raw.strip()
|
|
if name and name.lower() not in seen:
|
|
seen.add(name.lower())
|
|
context.append(name)
|
|
context = context[:10]
|
|
try:
|
|
limit = int(request.query.get("limit", 15))
|
|
except ValueError:
|
|
limit = 15
|
|
limit = max(1, min(limit, 50))
|
|
|
|
suggestions = await self._run_in_executor(
|
|
self.db.suggest_tags, prefix, context, limit
|
|
)
|
|
return web.json_response({"success": True, "suggestions": suggestions})
|
|
except Exception as e:
|
|
self.logger.error(f"Tag suggest error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": "Failed to suggest tags",
|
|
"suggestions": [],
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def get_tags_filter(self, request):
|
|
"""Get prompts matching multiple tags with AND/OR mode, or untagged prompts."""
|
|
try:
|
|
untagged = request.query.get("untagged", "").lower() == "true"
|
|
|
|
if untagged:
|
|
try:
|
|
limit, offset = parse_page_params(request.query, default_limit=20)
|
|
except ValueError:
|
|
return bad_request("limit and offset must be integers")
|
|
result = await self._run_in_executor(
|
|
self.db.get_untagged_prompts, limit, offset
|
|
)
|
|
self._present_prompts(result["prompts"])
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"tags": [],
|
|
"mode": "untagged",
|
|
"prompts": result["prompts"],
|
|
"pagination": {
|
|
"total": result["total"],
|
|
"limit": result["limit"],
|
|
"offset": result["offset"],
|
|
"has_more": result["has_more"],
|
|
},
|
|
}
|
|
)
|
|
|
|
tags_str = request.query.get("tags", "").strip()
|
|
if not tags_str:
|
|
return web.json_response(
|
|
{"success": False, "error": "Tags parameter required"}, status=400
|
|
)
|
|
|
|
tags_list = [t.strip() for t in tags_str.split(",") if t.strip()]
|
|
mode = request.query.get("mode", "and").lower()
|
|
if mode not in ("and", "or"):
|
|
mode = "and"
|
|
|
|
try:
|
|
limit, offset = parse_page_params(request.query, default_limit=20)
|
|
except ValueError:
|
|
return bad_request("limit and offset must be integers")
|
|
|
|
result = await self._run_in_executor(
|
|
self.db.get_prompts_by_tags, tags_list, mode, limit, offset
|
|
)
|
|
self._present_prompts(result["prompts"])
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"tags": tags_list,
|
|
"mode": mode,
|
|
"prompts": result["prompts"],
|
|
"pagination": {
|
|
"total": result["total"],
|
|
"limit": result["limit"],
|
|
"offset": result["offset"],
|
|
"has_more": result["has_more"],
|
|
},
|
|
}
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Tags filter error: {e}", exc_info=True)
|
|
return web.json_response(
|
|
{"success": False, "error": safe_error_message(e)}, status=500
|
|
)
|
|
|
|
async def rename_tag_endpoint(self, request):
|
|
"""Rename a tag across all prompts."""
|
|
try:
|
|
# aiohttp has already percent-decoded match_info; decoding again
|
|
# would turn a tag literally named "%41" into "A".
|
|
tag_name = request.match_info.get("tag_name", "")
|
|
if not tag_name:
|
|
return web.json_response(
|
|
{"success": False, "error": "Tag name required"}, status=400
|
|
)
|
|
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid JSON body"}, status=400
|
|
)
|
|
raw_name = body.get("new_name")
|
|
if not raw_name:
|
|
return web.json_response(
|
|
{"success": False, "error": "New tag name required"}, status=400
|
|
)
|
|
error = tags_error([raw_name])
|
|
if error is not None:
|
|
return error
|
|
new_name = raw_name.strip()
|
|
|
|
result = await self._run_in_executor(
|
|
self.db.rename_tag_all_prompts, tag_name, new_name
|
|
)
|
|
resp = {
|
|
"success": True,
|
|
"old_name": tag_name,
|
|
"new_name": new_name,
|
|
"affected_count": result["affected_count"],
|
|
}
|
|
if result.get("skipped_count", 0) > 0:
|
|
resp["skipped_count"] = result["skipped_count"]
|
|
resp["warning"] = (
|
|
f"{result['skipped_count']} prompt(s) had corrupted tag data "
|
|
"and were skipped"
|
|
)
|
|
return web.json_response(resp)
|
|
except Exception as e:
|
|
self.logger.error(f"Rename tag error: {e}", exc_info=True)
|
|
return web.json_response(
|
|
{"success": False, "error": safe_error_message(e)}, status=500
|
|
)
|
|
|
|
async def delete_tag_endpoint(self, request):
|
|
"""Delete a tag from all prompts."""
|
|
try:
|
|
# aiohttp has already percent-decoded match_info; decoding again
|
|
# would turn a tag literally named "%41" into "A".
|
|
tag_name = request.match_info.get("tag_name", "")
|
|
if not tag_name:
|
|
return web.json_response(
|
|
{"success": False, "error": "Tag name required"}, status=400
|
|
)
|
|
|
|
result = await self._run_in_executor(
|
|
self.db.delete_tag_all_prompts, tag_name
|
|
)
|
|
resp = {
|
|
"success": True,
|
|
"tag_name": tag_name,
|
|
"affected_count": result["affected_count"],
|
|
}
|
|
if result.get("skipped_count", 0) > 0:
|
|
resp["skipped_count"] = result["skipped_count"]
|
|
resp["warning"] = (
|
|
f"{result['skipped_count']} prompt(s) had corrupted tag data "
|
|
"and were skipped"
|
|
)
|
|
return web.json_response(resp)
|
|
except Exception as e:
|
|
self.logger.error(f"Delete tag error: {e}", exc_info=True)
|
|
return web.json_response(
|
|
{"success": False, "error": safe_error_message(e)}, status=500
|
|
)
|
|
|
|
async def merge_tags_endpoint(self, request):
|
|
"""Merge source tags into a target tag."""
|
|
try:
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid JSON body"}, status=400
|
|
)
|
|
source_tags = body.get("source_tags", [])
|
|
raw_target = body.get("target_tag")
|
|
|
|
if not source_tags:
|
|
return web.json_response(
|
|
{"success": False, "error": "Source tags required"}, status=400
|
|
)
|
|
if not raw_target:
|
|
return web.json_response(
|
|
{"success": False, "error": "Target tag required"}, status=400
|
|
)
|
|
if not isinstance(source_tags, list):
|
|
return bad_request("Source tags must be a list")
|
|
error = tags_error([*source_tags, raw_target])
|
|
if error is not None:
|
|
return error
|
|
target_tag = raw_target.strip()
|
|
|
|
result = await self._run_in_executor(
|
|
self.db.merge_tags, source_tags, target_tag
|
|
)
|
|
resp = {
|
|
"success": True,
|
|
"target_tag": target_tag,
|
|
"affected_count": result["affected_count"],
|
|
"tags_merged": result["tags_merged"],
|
|
}
|
|
if result.get("skipped_count", 0) > 0:
|
|
resp["skipped_count"] = result["skipped_count"]
|
|
resp["warning"] = (
|
|
f"{result['skipped_count']} prompt(s) had corrupted tag data "
|
|
"and were skipped"
|
|
)
|
|
return web.json_response(resp)
|
|
except Exception as e:
|
|
self.logger.error(f"Merge tags error: {e}", exc_info=True)
|
|
return web.json_response(
|
|
{"success": False, "error": safe_error_message(e)}, status=500
|
|
)
|
|
|
|
async def save_prompt(self, request):
|
|
"""Save a new prompt with metadata and duplicate detection."""
|
|
try:
|
|
data = await request.json()
|
|
|
|
text = (data.get("text") or "").strip()
|
|
if not text:
|
|
return web.json_response(
|
|
{"success": False, "error": "Text is required"}, status=400
|
|
)
|
|
|
|
category = (data.get("category") or "").strip() or None
|
|
tags = data.get("tags", [])
|
|
rating = data.get("rating") or None
|
|
notes = (data.get("notes") or "").strip() or None
|
|
|
|
try:
|
|
validate_prompt_text(text)
|
|
validate_category(category)
|
|
validate_tags(tags)
|
|
validate_rating(rating)
|
|
except ValueError as ve:
|
|
return web.json_response(
|
|
{"success": False, "error": str(ve)}, status=400
|
|
)
|
|
|
|
text = sanitize_input(text)
|
|
|
|
prompt_hash = generate_prompt_hash(text)
|
|
|
|
existing = await self._run_in_executor(
|
|
self.db.get_prompt_by_hash, prompt_hash
|
|
)
|
|
if existing:
|
|
if any([category, tags, rating, notes]):
|
|
await self._run_in_executor(
|
|
self.db.update_prompt_metadata,
|
|
prompt_id=existing["id"],
|
|
category=category,
|
|
tags=tags,
|
|
rating=rating,
|
|
notes=notes,
|
|
)
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"prompt_id": existing["id"],
|
|
"message": "Prompt already exists, metadata updated",
|
|
"is_duplicate": True,
|
|
}
|
|
)
|
|
|
|
prompt_id = await self._run_in_executor(
|
|
self.db.save_prompt,
|
|
text=text,
|
|
category=category,
|
|
tags=tags if tags else None,
|
|
rating=rating,
|
|
notes=notes,
|
|
prompt_hash=prompt_hash,
|
|
)
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"prompt_id": prompt_id,
|
|
"message": "Prompt saved successfully",
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Save error: {e}", exc_info=True)
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to save prompt: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def delete_prompt(self, request):
|
|
"""Delete a specific prompt by ID."""
|
|
try:
|
|
prompt_id = int(request.match_info["prompt_id"])
|
|
success = await self._run_in_executor(self.db.delete_prompt, prompt_id)
|
|
|
|
if success:
|
|
return web.json_response(
|
|
{"success": True, "message": "Prompt deleted successfully"}
|
|
)
|
|
else:
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": "Prompt not found or could not be deleted",
|
|
},
|
|
status=404,
|
|
)
|
|
|
|
except PairDependencyError as exc:
|
|
return web.json_response({"success": False, "error": str(exc)}, status=409)
|
|
except ValueError:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid prompt ID"}, status=400
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Delete error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to delete prompt: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def update_prompt(self, request):
|
|
"""Update prompt text."""
|
|
try:
|
|
prompt_id = int(request.match_info["prompt_id"])
|
|
data = await request.json()
|
|
new_text = (data.get("text") or "").strip()
|
|
|
|
if not new_text:
|
|
return web.json_response(
|
|
{"success": False, "error": "Text cannot be empty"}, status=400
|
|
)
|
|
|
|
try:
|
|
validate_prompt_text(new_text)
|
|
except ValueError as ve:
|
|
return web.json_response(
|
|
{"success": False, "error": str(ve)}, status=400
|
|
)
|
|
|
|
new_text = sanitize_input(new_text)
|
|
|
|
updated = await self._run_in_executor(
|
|
self.db.update_prompt_text, prompt_id, new_text
|
|
)
|
|
if updated:
|
|
return web.json_response(
|
|
{"success": True, "message": "Prompt updated successfully"}
|
|
)
|
|
else:
|
|
return web.json_response(
|
|
{"success": False, "error": "Prompt not found"}, status=404
|
|
)
|
|
|
|
except PairDependencyError as exc:
|
|
return web.json_response({"success": False, "error": str(exc)}, status=409)
|
|
except sqlite3.IntegrityError:
|
|
return web.json_response(
|
|
{"success": False, "error": "Another prompt already has this text"},
|
|
status=409,
|
|
)
|
|
except ValueError:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid prompt ID"}, status=400
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Update prompt error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to update prompt: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def update_prompt_rating(self, request):
|
|
"""Update prompt rating."""
|
|
try:
|
|
prompt_id = int(request.match_info["prompt_id"])
|
|
data = await request.json()
|
|
rating = data.get("rating")
|
|
|
|
try:
|
|
validate_rating(rating)
|
|
except ValueError as ve:
|
|
return web.json_response(
|
|
{"success": False, "error": str(ve)}, status=400
|
|
)
|
|
|
|
updated = await self._run_in_executor(
|
|
self.db.update_prompt_rating, prompt_id, rating
|
|
)
|
|
if updated:
|
|
return web.json_response(
|
|
{"success": True, "message": "Rating updated successfully"}
|
|
)
|
|
else:
|
|
return web.json_response(
|
|
{"success": False, "error": "Prompt not found"}, status=404
|
|
)
|
|
|
|
except ValueError:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid prompt ID"}, status=400
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Update rating error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to update rating: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def add_prompt_tag(self, request):
|
|
"""Add tag to prompt."""
|
|
try:
|
|
prompt_id = int(request.match_info["prompt_id"])
|
|
data = await request.json()
|
|
raw_tag = data.get("tag")
|
|
|
|
if not raw_tag:
|
|
return web.json_response(
|
|
{"success": False, "error": "Tag cannot be empty"}, status=400
|
|
)
|
|
error = tags_error([raw_tag])
|
|
if error is not None:
|
|
return error
|
|
new_tag = raw_tag.strip()
|
|
|
|
prompt = await self._run_in_executor(self.db.get_prompt_by_id, prompt_id)
|
|
if not prompt:
|
|
return web.json_response(
|
|
{"success": False, "error": "Prompt not found"}, status=404
|
|
)
|
|
|
|
current_tags = prompt.get("tags", [])
|
|
if not isinstance(current_tags, list):
|
|
current_tags = []
|
|
|
|
if new_tag not in current_tags:
|
|
current_tags.append(new_tag)
|
|
await self._run_in_executor(
|
|
self.db.set_prompt_tags, prompt_id, current_tags
|
|
)
|
|
|
|
return web.json_response(
|
|
{"success": True, "message": "Tag added successfully"}
|
|
)
|
|
|
|
except ValueError:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid prompt ID"}, status=400
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Add tag error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to add tag: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def add_tags_to_prompt(self, request):
|
|
"""Add multiple tags to a single prompt."""
|
|
try:
|
|
data = await request.json()
|
|
prompt_id = data.get("prompt_id")
|
|
new_tags = data.get("tags", [])
|
|
|
|
if not prompt_id:
|
|
return web.json_response(
|
|
{"success": False, "error": "Prompt ID is required"}, status=400
|
|
)
|
|
|
|
if not new_tags or not isinstance(new_tags, list):
|
|
return web.json_response(
|
|
{"success": False, "error": "Tags must be a non-empty list"},
|
|
status=400,
|
|
)
|
|
# Blank entries are skipped below, so only the rest is validated.
|
|
error = tags_error(
|
|
[t for t in new_tags if not (isinstance(t, str) and not t.strip())]
|
|
)
|
|
if error is not None:
|
|
return error
|
|
|
|
prompt = await self._run_in_executor(self.db.get_prompt_by_id, prompt_id)
|
|
if not prompt:
|
|
return web.json_response(
|
|
{"success": False, "error": "Prompt not found"}, status=404
|
|
)
|
|
|
|
current_tags = prompt.get("tags", [])
|
|
if not isinstance(current_tags, list):
|
|
current_tags = []
|
|
|
|
tags_added = 0
|
|
for new_tag in new_tags:
|
|
new_tag = new_tag.strip()
|
|
if new_tag and new_tag not in current_tags:
|
|
current_tags.append(new_tag)
|
|
tags_added += 1
|
|
|
|
if tags_added > 0:
|
|
await self._run_in_executor(
|
|
self.db.set_prompt_tags, prompt_id, current_tags
|
|
)
|
|
|
|
message = f"{tags_added} tag(s) added successfully"
|
|
if tags_added == 0:
|
|
message = "No new tags to add (all tags already exist)"
|
|
|
|
return web.json_response(
|
|
{"success": True, "message": message, "tags_added": tags_added}
|
|
)
|
|
|
|
except ValueError:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid prompt ID"}, status=400
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Add tags error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to add tags: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def remove_prompt_tag(self, request):
|
|
"""Remove tag from prompt."""
|
|
try:
|
|
prompt_id = int(request.match_info["prompt_id"])
|
|
data = await request.json()
|
|
tag_to_remove = (data.get("tag") or "").strip()
|
|
|
|
prompt = await self._run_in_executor(self.db.get_prompt_by_id, prompt_id)
|
|
if not prompt:
|
|
return web.json_response(
|
|
{"success": False, "error": "Prompt not found"}, status=404
|
|
)
|
|
|
|
current_tags = prompt.get("tags", [])
|
|
if not isinstance(current_tags, list):
|
|
current_tags = []
|
|
|
|
if tag_to_remove in current_tags:
|
|
current_tags.remove(tag_to_remove)
|
|
await self._run_in_executor(
|
|
self.db.set_prompt_tags, prompt_id, current_tags
|
|
)
|
|
|
|
return web.json_response(
|
|
{"success": True, "message": "Tag removed successfully"}
|
|
)
|
|
|
|
except ValueError:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid prompt ID"}, status=400
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(f"Remove tag error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to remove tag: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def bulk_delete_prompts(self, request):
|
|
"""Bulk delete prompts."""
|
|
try:
|
|
data = await request.json()
|
|
prompt_ids = data.get("prompt_ids", [])
|
|
|
|
if not prompt_ids:
|
|
return web.json_response(
|
|
{"success": False, "error": "No prompt IDs provided"}, status=400
|
|
)
|
|
|
|
deleted_count = await self._run_in_executor(
|
|
self.db.bulk_delete_prompts, prompt_ids
|
|
)
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"message": f"Deleted {deleted_count} prompts",
|
|
"deleted_count": deleted_count,
|
|
}
|
|
)
|
|
|
|
except PairDependencyError as exc:
|
|
return web.json_response({"success": False, "error": str(exc)}, status=409)
|
|
except Exception as e:
|
|
self.logger.error(f"Bulk delete error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to delete prompts: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def bulk_add_tags(self, request):
|
|
"""Bulk add tags to prompts."""
|
|
try:
|
|
data = await request.json()
|
|
prompt_ids = data.get("prompt_ids", [])
|
|
new_tags = data.get("tags", [])
|
|
|
|
if not prompt_ids or not new_tags:
|
|
return web.json_response(
|
|
{"success": False, "error": "No prompt IDs or tags provided"},
|
|
status=400,
|
|
)
|
|
if not isinstance(new_tags, list):
|
|
return bad_request("Tags must be a list")
|
|
error = tags_error(new_tags)
|
|
if error is not None:
|
|
return error
|
|
|
|
updated_count = await self._run_in_executor(
|
|
self.db.bulk_add_tags, prompt_ids, new_tags
|
|
)
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"message": f"Added tags to {updated_count} prompts",
|
|
"updated_count": updated_count,
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Bulk add tags error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to add tags: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
async def bulk_set_category(self, request):
|
|
"""Bulk set category for prompts."""
|
|
try:
|
|
try:
|
|
data = await request.json()
|
|
except ValueError:
|
|
return web.json_response(
|
|
{"success": False, "error": "Invalid JSON body"}, status=400
|
|
)
|
|
if not isinstance(data, dict):
|
|
return web.json_response(
|
|
{"success": False, "error": "Expected a JSON object"}, status=400
|
|
)
|
|
if not data.get("prompt_ids"):
|
|
return web.json_response(
|
|
{"success": False, "error": "No prompt IDs provided"}, status=400
|
|
)
|
|
category = data.get("category")
|
|
try:
|
|
prompt_ids = parse_bulk_prompt_ids(data["prompt_ids"])
|
|
validate_category(category)
|
|
except ValueError as ve:
|
|
return web.json_response(
|
|
{"success": False, "error": str(ve)}, status=400
|
|
)
|
|
category = (category or "").strip()
|
|
|
|
updated_count = await self._run_in_executor(
|
|
self.db.bulk_set_category, prompt_ids, category
|
|
)
|
|
|
|
return web.json_response(
|
|
{
|
|
"success": True,
|
|
"message": f"Set category for {updated_count} prompts",
|
|
"updated_count": updated_count,
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Bulk set category error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to set category: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|
|
|
|
def _all_prompts_for_export(self):
|
|
"""Every prompt, fetched in EXPORT_PAGE_SIZE pages (blocking)."""
|
|
prompts, offset = [], 0
|
|
while True:
|
|
page = self.db.search_prompts(limit=EXPORT_PAGE_SIZE, offset=offset)
|
|
prompts.extend(page)
|
|
if len(page) < EXPORT_PAGE_SIZE:
|
|
return prompts
|
|
offset += EXPORT_PAGE_SIZE
|
|
|
|
async def export_prompts(self, request):
|
|
"""Export all prompts to JSON."""
|
|
try:
|
|
prompts = await self._run_in_executor(self._all_prompts_for_export)
|
|
|
|
export_data = {
|
|
"export_date": datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
"total_prompts": len(prompts),
|
|
"prompts": prompts,
|
|
}
|
|
|
|
json_data = json.dumps(export_data, indent=2, ensure_ascii=False)
|
|
|
|
return web.Response(
|
|
text=json_data,
|
|
content_type="application/json",
|
|
headers={
|
|
"Content-Disposition": (
|
|
'attachment; filename="prompt_manager_'
|
|
f'{datetime.datetime.now().strftime("%Y%m%d_%H%M%S")}.json"'
|
|
)
|
|
},
|
|
)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Export error: {e}")
|
|
return web.json_response(
|
|
{
|
|
"success": False,
|
|
"error": f"Failed to export prompts: {safe_error_message(e)}",
|
|
},
|
|
status=500,
|
|
)
|