Vito
caa9649f32
* feat(db): add pair schema and stage restore migrations for #193 * feat(database): persist reusable ordered prompt pairs (#193) * feat(api): expose reusable pairs and atomic metadata enrichment (#193) * feat(metadata): resolve canonical prompt pairs with bounded provenance (#193) * feat(scan): enrich image pairs without guessing or duplicate links (#193) * fix(database): preserve pair tags during global maintenance (#193) * feat(tracking): enrich live image pairs and cancel scans safely (#193) * feat(nodes): encode paired prompts and count cached recipe runs (#193) * fix(database): validate restored pair ownership and provenance (#193) * feat(admin): report pair scan outcomes and cancellation (#193) * feat(ui): add paired prompt editor and explicit recipe library for #193 * fix(autocomplete): include reusable pair metadata vocabulary (#193) * fix(pairs): honor recent sorting and conservative scan outcomes (#193) * fix(ui): protect paired draft saves and clarify negative variants (#193) * fix(metadata): preserve bounded paired execution evidence (#193) * fix(tracking): follow actual pair output roles and prefer execution evidence (#193) * docs(pairs): document live validation and retain renderer screenshots (#193) * test(api): inject disposable databases before API startup (#194) * test(api): canonicalize portable temporary gallery roots (#194) * docs(ui): prototype branching negative variant history (#193) * docs(ui): show selected negative beneath positive in timeline concept (#193) * docs: retain approved timeline for later gallery redesign (#193) * test(db): close restore fixture connections before cleanup (#194) * test(db): reuse existing handle guards in all pair fixtures (#194) * chore: clarify rollback test and intentional exception handling (#193)
227 lines
8.8 KiB
Python
227 lines
8.8 KiB
Python
"""Paired positive and negative conditioning with reusable recipe storage."""
|
|
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from typing import Any, Dict, Optional, Tuple
|
|
|
|
try:
|
|
from .prompt_manager import IO, InputTypeDict, PromptManagerFocused
|
|
from .utils.hashing import generate_prompt_hash
|
|
from .utils.pair_execution import record_pair_execution
|
|
from .utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
|
|
from .utils.prompt_pair_resolver import resolve_text_input
|
|
from .utils.usage_tracking import (
|
|
PENDING_FIRST_USE,
|
|
PENDING_PAIR_FIRST_USE,
|
|
pair_run_key,
|
|
)
|
|
except ImportError:
|
|
from prompt_manager import IO, InputTypeDict, PromptManagerFocused
|
|
from utils.hashing import generate_prompt_hash
|
|
from utils.pair_execution import record_pair_execution
|
|
from utils.prompt_graph import intrinsic_pair_nodes, run_prompt_components
|
|
from utils.prompt_pair_resolver import resolve_text_input
|
|
from utils.usage_tracking import (
|
|
PENDING_FIRST_USE,
|
|
PENDING_PAIR_FIRST_USE,
|
|
pair_run_key,
|
|
)
|
|
|
|
|
|
class PromptManagerPair(PromptManagerFocused):
|
|
"""Keep raw editors as one recipe while encoding their effective texts."""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls) -> InputTypeDict:
|
|
inputs = super().INPUT_TYPES()
|
|
inputs["required"] = {
|
|
"text": inputs["required"]["text"],
|
|
"negative_text": (
|
|
IO.STRING,
|
|
{"default": "", "multiline": True, "dynamicPrompts": True},
|
|
),
|
|
"clip": inputs["required"]["clip"],
|
|
}
|
|
inputs["hidden"]["extra_pnginfo"] = "EXTRA_PNGINFO"
|
|
return inputs
|
|
|
|
RETURN_TYPES = (IO.CONDITIONING, IO.CONDITIONING, IO.STRING, IO.STRING)
|
|
RETURN_NAMES = ("positive", "negative", "positive_text", "negative_text")
|
|
OUTPUT_TOOLTIPS = (
|
|
"Positive conditioning with composition and optional LoRA trigger words.",
|
|
"Negative conditioning from the negative editor, including empty text.",
|
|
"The effective positive text that was encoded.",
|
|
"The unchanged negative text that was encoded.",
|
|
)
|
|
DESCRIPTION = "Encode and save a reusable positive/negative prompt pair."
|
|
|
|
def encode_prompt(
|
|
self,
|
|
clip: Any,
|
|
text: str,
|
|
negative_text: str = "",
|
|
category: str = "",
|
|
tags: str = "",
|
|
search_text: str = "",
|
|
prepend_text: str = "",
|
|
append_text: str = "",
|
|
prompt: Optional[Dict[str, Any]] = None,
|
|
unique_id: Optional[str] = None,
|
|
extra_pnginfo: Optional[Dict[str, Any]] = None,
|
|
) -> Tuple[Any, Any, str, str]:
|
|
"""Encode both sides once; database and tracking failures never block CLIP."""
|
|
if clip is None:
|
|
raise RuntimeError("ERROR: clip input is invalid: None")
|
|
if not isinstance(text, str) or not isinstance(negative_text, str):
|
|
raise ValueError("Positive and negative prompt text must be strings")
|
|
positive = self._inject_lora_trigger_words(
|
|
" ".join(
|
|
part
|
|
for part in (prepend_text.strip(), text, append_text.strip())
|
|
if part
|
|
)
|
|
)
|
|
positive_conditioning = clip.encode_from_tokens_scheduled(
|
|
clip.tokenize(positive)
|
|
)
|
|
negative_conditioning = clip.encode_from_tokens_scheduled(
|
|
clip.tokenize(negative_text)
|
|
)
|
|
record_pair_execution(
|
|
extra_pnginfo,
|
|
unique_id,
|
|
text=text,
|
|
negative_text=negative_text,
|
|
positive_text=positive,
|
|
negative_effective_text=negative_text,
|
|
prepend_text=prepend_text.strip(),
|
|
append_text=append_text.strip(),
|
|
)
|
|
pair_id = prompt_id = None
|
|
known = isinstance(prompt, dict) and unique_id is not None
|
|
node_id = str(unique_id)
|
|
used_fields = (
|
|
{
|
|
field
|
|
for source, field in run_prompt_components(prompt)
|
|
if source == node_id
|
|
}
|
|
if known
|
|
else {"text"}
|
|
)
|
|
is_positive = "text" in used_fields
|
|
pair_used = node_id in intrinsic_pair_nodes(prompt) if known else True
|
|
try:
|
|
pair_id = self.db.save_prompt_pair(
|
|
text,
|
|
negative_text,
|
|
category=category.strip() or None,
|
|
tags=self._parse_tags(tags),
|
|
)
|
|
pair = self.db.get_prompt_pair(pair_id)
|
|
prompt_id = pair["positive_prompt_id"]
|
|
resolved_positive = (
|
|
resolve_text_input(prompt, node_id, "text") if known else None
|
|
)
|
|
resolved_negative = (
|
|
resolve_text_input(prompt, node_id, "negative_text") if known else None
|
|
)
|
|
positive_seen = resolved_positive is not None and generate_prompt_hash(
|
|
resolved_positive
|
|
) == generate_prompt_hash(text)
|
|
pair_seen = (
|
|
positive_seen
|
|
and resolved_negative is not None
|
|
and generate_prompt_hash(resolved_negative)
|
|
== generate_prompt_hash(negative_text)
|
|
)
|
|
positive_pending = getattr(self, "pending_first_use", PENDING_FIRST_USE)
|
|
pair_pending = getattr(
|
|
self, "pending_pair_first_use", PENDING_PAIR_FIRST_USE
|
|
)
|
|
counted = set()
|
|
for field, component_text, resolved_text, component_id in (
|
|
("text", text, resolved_positive, prompt_id),
|
|
(
|
|
"negative_text",
|
|
negative_text,
|
|
resolved_negative,
|
|
pair["negative_prompt_id"],
|
|
),
|
|
):
|
|
component_hash = generate_prompt_hash(component_text)
|
|
if (
|
|
field not in used_fields
|
|
or not component_id
|
|
or component_hash in counted
|
|
):
|
|
continue
|
|
counted.add(component_hash)
|
|
seen_at_queue = (
|
|
resolved_text is not None
|
|
and generate_prompt_hash(resolved_text) == component_hash
|
|
)
|
|
runs = positive_pending.consume(component_hash) + int(not seen_at_queue)
|
|
if runs:
|
|
self.db.record_prompt_use(component_id, times=runs)
|
|
if pair_used:
|
|
runs = pair_pending.consume(pair_run_key(text, negative_text)) + int(
|
|
not pair_seen
|
|
)
|
|
if runs:
|
|
self.db.record_prompt_pair_use(pair_id, times=runs)
|
|
if is_positive:
|
|
self.prompt_tracker.set_current_prompt(
|
|
prompt_text=positive,
|
|
additional_data={
|
|
"prompt_id": prompt_id,
|
|
"pair_id": pair_id if pair_used else None,
|
|
"positive_text": positive,
|
|
"negative_text": negative_text,
|
|
"positive_storage_text": text.strip(),
|
|
"negative_storage_text": negative_text.strip(),
|
|
"prepend_text": prepend_text.strip(),
|
|
"append_text": append_text.strip(),
|
|
"final_text": positive,
|
|
"category": category.strip() or None,
|
|
"tags": self._parse_tags(tags) or [],
|
|
},
|
|
push_to_queue=not positive_seen or (pair_used and not pair_seen),
|
|
)
|
|
except Exception as error:
|
|
self.logger.warning(f"Could not save or track paired prompt: {error}")
|
|
try:
|
|
self.comfyui_integration.register_prompt(
|
|
f"promptmanager_pair_{time.time_ns()}",
|
|
positive,
|
|
{
|
|
"prompt_id": prompt_id,
|
|
"pair_id": pair_id if pair_used else None,
|
|
"negative_text": negative_text,
|
|
"category": category.strip() or None,
|
|
},
|
|
)
|
|
except Exception as error:
|
|
self.logger.warning(f"Could not register paired metadata: {error}")
|
|
return positive_conditioning, negative_conditioning, positive, negative_text
|
|
|
|
@classmethod
|
|
def IS_CHANGED(
|
|
cls,
|
|
clip: Any,
|
|
text: str = "",
|
|
negative_text: str = "",
|
|
category: str = "",
|
|
tags: str = "",
|
|
search_text: str = "",
|
|
prepend_text: str = "",
|
|
append_text: str = "",
|
|
**kwargs: Any,
|
|
) -> str:
|
|
"""Either editor, composition or pair metadata invalidates the node cache."""
|
|
serialized = json.dumps(
|
|
[text, negative_text, prepend_text, append_text, category, tags]
|
|
)
|
|
return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
|