import random import hashlib import json as _json import os as _os from copy import deepcopy from src.prompt_parser import ParsedPrompt from src.tag_warehouse import TagWarehouse, MAX_RATING_MAP from src.model_formatter import format_prompt, _is_score_tag from src.tag_format import to_internal_tag, normalize_tag from src.synonym_filter import apply_synonym_filter, has_synonym_conflict from src.prompt_rewriter import get_tag_categories from src.presets import PRESETS, get_preset_bundle_tags from src.dedup_engine import smart_dedup from src.synonym_data import _find_synonym_group, get_synonym_groups from src.tag_searcher import has_human_subject, filter_subject_conflicts, get_cooccurrence_tags from src.semantic_coherence import ( semantic_pick_tags, split_core_decorative, detect_intent, compute_theme_budget, _THEME_GROUPS, SCORING, ) CREATIVITY_SETTINGS = { "very_low": { "tags_per_category": (1, 1), "replacement_rate": 0.1, "shuffle_general": False, "concept_cross": False, "diversity_threshold": 0.5, "wildcard_categories": 0, "substitution_chance": 0.0, }, "low": { "tags_per_category": (1, 2), "replacement_rate": 0.2, "shuffle_general": False, "concept_cross": False, "diversity_threshold": 0.4, "wildcard_categories": 0, "substitution_chance": 0.05, }, "medium": { "tags_per_category": (2, 4), "replacement_rate": 0.3, "shuffle_general": False, "concept_cross": False, "diversity_threshold": 0.3, "wildcard_categories": 0, "substitution_chance": 0.1, }, "high": { "tags_per_category": (3, 6), "replacement_rate": 0.4, "shuffle_general": True, "concept_cross": True, "diversity_threshold": 0.25, "wildcard_categories": 1, "substitution_chance": 0.15, }, "very_high": { "tags_per_category": (4, 8), "replacement_rate": 0.5, "shuffle_general": True, "concept_cross": True, "diversity_threshold": 0.2, "wildcard_categories": 2, "substitution_chance": 0.2, }, "extreme": { "tags_per_category": (5, 10), "replacement_rate": 0.6, "shuffle_general": True, "concept_cross": True, "diversity_threshold": 0.15, "wildcard_categories": 3, "substitution_chance": 0.3, }, } CONFLICT_GROUPS = [ {"from above", "from below", "from side", "from behind", "birds-eye view", "worms-eye view"}, {"smile", "laughing", "serious", "angry", "sad", "crying", "surprised", "expressionless"}, {"standing", "sitting", "lying", "kneeling", "running", "jumping", "dancing", "crouching"}, {"simple background", "detailed background", "complex background", "gradient background"}, {"day", "night", "twilight", "sunset", "sunrise"}, {"portrait", "full body", "upper body", "lower body", "cowboy shot", "wide shot"}, {"soft lighting", "hard lighting", "harsh lighting"}, {"warm colors", "cool colors", "monochrome", "vibrant colors", "pastel colors", "dark colors", "neon palette"}, {"looking at viewer", "looking away", "looking up", "looking down", "looking back"}, {"sunlight", "moonlight"}, {"cloudy sky", "starry sky", "clear sky"}, {"misty", "foggy"}, {"indoors", "outdoors"}, {"innocent", "seductive smile"}, {"rainy", "snowy"}, {"close-up", "wide shot", "extreme close-up"}, {"chibi", "photorealistic"}, {"bloom", "soft focus"}, {"underwater", "space", "cityscape"}, {"hoodie", "poncho", "t-shirt", "crop top", "sweater", "jacket", "coat", "blazer", "vest", "cardigan", "blouse", "shirt", "tank top", "cape", "cloak", "robe", "dress", "kimono", "yukata", "qipao", "overalls", "suit", "armor"}, {"jeans", "shorts", "skirt", "mini skirt", "leggings", "sweatpants", "trousers", "hot pants", "long skirt", "pleated skirt"}, {"shoes", "boots", "sandals", "sneakers", "heels", "loafers", "flip-flops", "socks", "thighhighs", "kneehighs"}, {"hat", "cap", "beanie", "crown", "hood", "veil", "witch hat", "tara", "hairband", "headband", "maid headdress", "cat ears", "fox ears", "animal ears"}, {"confident", "shy", "timid", "bashful", "embarrassed", "proud", "self-assured"}, ] _CONFLICT_INDEX = None _CONFLICT_ONLY_INDEX = None def _build_conflict_index() -> dict[str, set]: global _CONFLICT_INDEX global _CONFLICT_ONLY_INDEX index: dict[str, set] = {} only: dict[str, set] = {} for group in CONFLICT_GROUPS: for member in group: ml = member.lower().strip() index[ml] = group only[ml] = group for group in get_synonym_groups(): for member in group: ml = member.lower().strip() if ml in index: index[ml] = index[ml] | group else: index[ml] = group _CONFLICT_INDEX = index _CONFLICT_ONLY_INDEX = only return _CONFLICT_INDEX def _find_conflict_group(tag: str) -> set | None: global _CONFLICT_INDEX if _CONFLICT_INDEX is None: _build_conflict_index() return _CONFLICT_INDEX.get(tag.lower().strip()) def _find_conflict_only_group(tag: str) -> set | None: """Conflict group WITHOUT synonym members. Used for negative-prompt inversion: a synonym of a desired tag shares its meaning, so it must NOT be negated (that would fight the positive prompt). """ global _CONFLICT_ONLY_INDEX if _CONFLICT_ONLY_INDEX is None: _build_conflict_index() return _CONFLICT_ONLY_INDEX.get(tag.lower().strip()) def _find_conflicts(tag: str, existing: list[str]) -> set[str]: group = _find_conflict_group(tag) if group is None: return set() existing_lower = {e.lower().strip() for e in existing} conflicting = group & existing_lower return {e for e in existing if e.lower().strip() in conflicting} def _resolve_and_replace(tag: str, existing: list[str], warehouse=None, protected=None) -> tuple[list[str], bool]: tl = tag.lower().strip() protected = protected or set() existing_lower = {e.lower().strip() for e in existing} conflicting = _find_conflicts(tag, existing) # Exclude the tag itself (case-insensitive dup) from conflict resolution. conflicting = {e for e in conflicting if e.lower().strip() != tl} if warehouse: for e in existing: if warehouse.has_conflict(tag, [e]): conflicting.add(e) # Never remove a protected (user/core) tag. If the only conflict is protected, # drop the incoming tag instead so user intent is always preserved. for e in conflicting: if e.lower().strip() in protected: return existing, False already = tl in existing_lower if not conflicting and already: return existing, False out = [e for e in existing if e not in conflicting] if not already: out.append(tag) return out, not already def _substitute_at(result: list[str], i: int, best: str, warehouse=None, protected=None) -> list[str]: """Replace result[i] with best, preserving conflict removal from _resolve_and_replace. _resolve_and_replace returns the full list with conflicting tags removed; the substitution caller must use that list (not just overwrite position i) or the removed conflict would silently reappear at its original position. """ other = [t for j, t in enumerate(result) if j != i] out, ok = _resolve_and_replace(best, other, warehouse, protected) if not ok: return result new_result = [] for j, t in enumerate(result): if j == i: new_result.append(best) elif t in out: new_result.append(t) return new_result def _extract_artists_from_general(variant, known_artists: dict[str, str]) -> None: """Move tags that match known artist names into ``variant.artists``. Operates in place on ``variant.general_tags``. ``known_artists`` MUST be built once per generation call (it is independent of the variation loop). """ skip = { (variant.subject or "").lower().strip(), (variant.character or "").lower().strip(), (variant.series or "").lower().strip(), } skip.discard("") remaining = [] for tag in variant.general_tags: tl = tag.lower().strip() if tl in skip: continue matched = known_artists.get(tl) if matched and matched not in variant.artists: variant.artists.append(matched) else: remaining.append(tag) variant.general_tags = remaining def _min_diversity_index(tags1: list[str], tags2: list[str]) -> float: if not tags1 or not tags2: return 1.0 s1 = {t.lower().strip() for t in tags1} s2 = {t.lower().strip() for t in tags2} diff = (s1 | s2) - (s1 & s2) return len(diff) / max(len(s1 | s2), 1) # --- Smart balancing helpers (booru best practices) --- # Restrictive metadata tags stripped from positive prompts # Auto-stripped "noise / restrictive / metadata" tags (mirrors Booru Prompt # Gallery's "removes metadata / restrictive tags" behavior + our extensions). RESTRICTIVE_TAGS = frozenset({ "white background", "simple background", "black background", "transparent background", "grey background", "gray background", "censor", "censored", "mosaic censorship", "censorship", "web address", "patreon logo", "patreon username", "commentary request", "translated", "english text", # --- extended metadata / artifact strip (C) --- "watermark", "artist name", "signature", "edited", "edit", "cropped", "crop", "duplicate", "disapproved", "resized", "compressed", "real life", "photograph", "photo", "raw image", "source image", "screencap", "screenshot", "scanned", "recycled", "upscaled", "downscaled", "downsampling", "high resolution", "low resolution", "4k", "8k", "12k", "deformed", "bad art", "worst quality", "low quality", }) # Canonical alias map (Booru Tag Gallery "Alias Resolver", G). Maps common # non-canonical spellings to the booru-standard tag. Applied during cleaning. # NOTE: Danbooru canon is `blonde hair` (NOT `yellow hair`) — that is a common # heuristic mistake; aliases below only canonicalize well-known, unambiguous # spellings, and never overwrite an official booru tag with a non-booru one. _BOORU_ALIASES = { "blond hair": "blonde hair", "blonde hair": "blonde hair", "blonde_hair": "blonde hair", "blond": "blonde hair", "grey hair": "gray hair", "grey eyes": "gray eyes", "grey background": "gray background", "grey skin": "gray skin", "platinum hair": "white hair", "silver hair": "white hair", "violet hair": "purple hair", "lavender hair": "purple hair", "scarlet hair": "red hair", "crimson hair": "red hair", "azure hair": "blue hair", "light blue hair": "aqua hair", "cyan hair": "aqua hair", "turquoise hair": "aqua hair", "magenta hair": "pink hair", "rose hair": "pink hair", "pink-colored": "pink", "coloured skin": "colored skin", "colored eyes": "colored pupils", "heterochromia eyes": "heterochromia", "mismatched eyes": "heterochromia", "two-tone hair": "multicolored hair", "two toned hair": "multicolored hair", "gradient hair": "multicolored hair", "ombre hair": "multicolored hair", "multicolor eyes": "heterochromia", "expressionless face": "expressionless", "looking back": "looking at viewer", # canonical spike "looking_at_viewer": "looking at viewer", # pose tagger outputs underscore "looking_away": "looking away", # pose tagger outputs underscore "looking_up": "looking up", "looking_down": "looking down", "looking_back": "looking at viewer", "closed_eyes": "eyes closed", "arms_up": "arms up", "arms_behind_head": "arms behind head", "arms_at_sides": "arms at sides", "arms_crossed": "arms crossed", "arms_behind_back": "arms behind back", "arms_around_neck": "arms around neck", "hands_on_hips": "hands on hips", "hand_on_hip": "hand on hip", "hand_on_own_face": "hand on own face", "hand_on_own_chest": "hand on own chest", "hand_on_own_stomach": "hand on own stomach", "hand_in_pocket": "hand in pocket", "hands_in_pockets": "hands in pockets", "covering_face": "covering face", "head_tilt": "head tilt", "one_arm_up": "one arm up", "one_leg_up": "one leg up", "standing_on_one_leg": "standing on one leg", "on_all_fours": "on all fours", "kneeling_on_one_knee": "kneeling on one knee", "crossed_legs": "crossed legs", "legs_apart": "legs apart", "close up": "close-up", "extreme close up": "extreme close-up", "upper body": "upper body", "lower body": "lower body", "full body": "full body", "cowboy shot": "cowboy shot", "dutch angle": "dutch angle", "birds eye view": "bird's-eye view", "birds-eye view": "bird's-eye view", "worms eye view": "worm's-eye view", "worms-eye view": "worm's-eye view", "white background": "white background", "simple background": "simple background", "gradient background": "gradient background", "detailed background": "detailed background", "blurry background": "blurry background", "blurred background": "blurry background", "starry sky": "starry sky", "cloudy sky": "cloudy sky", "clear sky": "clear sky", "night sky": "night sky", "starry sky": "starry sky", # canonical "sunny": "sunny", "rainy": "rainy", "snowy": "snowy", "foggy": "foggy", "misty": "foggy", "overcast": "cloudy sky", "golden hour": "golden hour", "blue hour": "dusk", "twilight": "dusk", "dawn": "sunrise", "golden lighting": "golden hour", "dramatic lighting": "dramatic lighting", "cinematic lighting": "cinematic lighting", "volumetric lighting": "volumetric lighting", "soft lighting": "soft lighting", "hard lighting": "hard lighting", "harsh lighting": "hard lighting", "studio lighting": "studio lighting", "natural lighting": "natural lighting", "neon lighting": "neon lighting", "glowing": "glowing", "lens flare": "lens flare", "bloom": "bloom", "depth of field": "depth of field", "bokeh": "bokeh", "chromatic aberration": "chromatic aberration", "film grain": "film grain", "vignette": "vignette", "motion blur": "motion blur", "sharp focus": "sharp focus", "soft focus": "soft focus", } # Token ceiling for one variation (Anima/Illustrious sweet spot ~75). MAX_TOKENS_DEFAULT = 75 FX_CATEGORIES = ["effects", "special_fx", "lighting", "atmosphere"] # Negative-quality / source-metadata tags that must never land in a positive prompt. _NEG_QUALITY_BAD = frozenset({ "worst quality", "bad quality", "low quality", "worst aesthetic", "worst score", "low score", "average score", "displeasing", "very displeasing", "bad aesthetic", "normal quality", }) def _is_negative_quality(tag: str) -> bool: tl = tag.lower().strip() if tl in _NEG_QUALITY_BAD: return True if "displeasing" in tl: return True if tl.startswith("worst") or tl.startswith("low ") or tl.startswith("bad "): return True if tl.startswith("score_"): try: num = int(tl.split("_", 1)[1]) except (ValueError, IndexError): return False return num < 7 if tl.startswith("source_"): return True return False # Categories whose tags describe the subject's fixed design (identity) rather # than flexible scene/ambiance. Used by Full Rewrite to keep the concept while # rebuilding everything else. _DESIGN_CATS = { "quality", "expression", "hair", "body", "pose", "clothing", "background", "lighting", "effects", "atmosphere", "colors", "composition", "style", } # Semantic role priority for final tag ordering (lower = earlier in prompt). # SD gives earlier tokens more attention, so identity/composition must lead # and ambiance/effects must trail; unknown categories stay in a stable tail. _TAG_ORDER_PRIORITY = { "quality": 0, "year_meta": 0, "composition": 1, "framing": 1, "pose": 2, "expression": 2, "body": 3, "hair": 3, "eyes": 3, "colors": 3, "face": 3, "makeup": 3, "accessory": 4, "clothing": 4, "background": 5, "architecture": 5, "season": 5, "atmosphere": 5, "weather": 5, "lighting": 6, "color_grading": 6, "effects": 7, "special_fx": 7, "style": 8, "bloom": 8, "object": 9, "food": 9, "weapon": 9, "vehicle": 9, "demon": 10, "angelic": 10, "magic": 10, "animal": 11, "furry": 11, "nsfw": 12, "horror": 13, "fantasy": 13, "cyberpunk": 13, "gothic": 13, "steampunk": 13, "noir": 13, "retro": 13, "kawaii": 13, "watercolor": 13, "space": 13, "underwater": 13, "warrior": 13, "magical_girl": 13, } def _smart_order_tags(tags: list[str]) -> list[str]: """Stable-sort tags by semantic role so identity/composition lead the prompt. Stable sort preserves the existing order inside each role group (e.g. the order the user wrote their hair/eye tags). Tags with no known category (subject names like ``1girl``, character tags) rank at the front so user intent keeps attention; genuinely unknown additions fall to the tail. """ def _rank(t: str) -> int: cats = get_tag_categories(t) if not cats: return 1 return min(_TAG_ORDER_PRIORITY.get(c, 50) for c in cats) return sorted(tags, key=_rank) def _protected_set(parsed: ParsedPrompt, design_only: bool = False) -> set[str]: """Tags that must never be removed or overwritten by conflict resolution / budgeting. design_only=False (Standard Varry): protect the user's subject, character, series, artists, quality tags AND every explicit general tag they wrote, so variations only ADD variety and never erase the user's intent. design_only=True (Full Rewrite): protect subject/character/series + only the design-category general tags, letting the rest of the prompt be rebuilt. """ p: set[str] = set() if parsed.subject: p.add(parsed.subject.lower().strip()) if parsed.character: for c in parsed.character.split(","): c = c.strip() if c: p.add(c.lower()) if parsed.series: p.add(parsed.series.lower().strip()) for a in (parsed.artists or []): p.add(a.lower().strip()) for q in (parsed.quality_tags or []): p.add(q.lower().strip()) for t in (parsed.general_tags or []): tl = t.lower().strip() if design_only: cats = get_tag_categories(t) if any(c in _DESIGN_CATS for c in cats): p.add(tl) else: p.add(tl) return p def _head(tag: str) -> str: """Last whitespace-separated token of a tag (its head noun), lower-cased.""" parts = tag.split() return parts[-1].lower() if parts else tag.lower() import re as _re _JUNK_RE = _re.compile(r"(//|cid=|[<>]|[\"'`])") def _clean_general_tags(tags: list[str]) -> list[str]: """Strip crawler/UI artifacts and normalize spaced score tags. e.g. '0.4//cid=12>' -> dropped, 'score 9' -> 'score_9'. Keeps the user's meaningful tags intact so downstream generation stays clean. """ out = [] for t in tags: tl = t.strip() if not tl: continue m = _re.fullmatch(r"score\s+(\d+)", tl, _re.I) if m: out.append(f"score_{m.group(1)}") continue if _re.fullmatch(r"\d+(\.\d+)?", tl): continue if _JUNK_RE.search(tl): continue # Normalize booru underscores to the spaced canonical form (except # score_N) so category / co-occurrence / conflict lookups match and so # user tags and injected tags share one consistent surface form. This # is what prevents mixed "looking_at_viewer, magical girl pose" output. tl = to_internal_tag(tl) # Alias Resolver (G): canonicalize common non-standard spellings. alias = _BOORU_ALIASES.get(tl.lower()) if alias: tl = alias out.append(tl) return out def _is_known_tag(tag: str) -> bool: """True if the tag is a recognized pool/category tag (vs a user-specific name).""" return bool(get_tag_categories(tag)) def _normalize_exclude(exclude_tags: list[str] | None) -> set[str]: if not exclude_tags: return set() return {t.lower().strip() for t in exclude_tags if t and t.strip()} def _apply_strip_flags( parsed: "ParsedPrompt", warehouse: "TagWarehouse", strip_quality: bool = False, strip_artist: bool = False, strip_lora: bool = False, strip_meta: bool = False, ) -> "ParsedPrompt": """Apply the B (cleanup toggles) flags to a parsed prompt in place. - strip_quality: drop quality tokens from general + quality_tags - strip_artist: drop artist entries (so they are not protected either) - strip_lora: drop LoRA/embed trigger tokens (':', '<', '>') - strip_meta: drop meta tokens """ if strip_quality: parsed.quality_tags = [] parsed.general_tags = [ t for t in parsed.general_tags if not ("quality" in t.lower() or "masterpiece" in t.lower() or _is_score_tag(t)) ] if strip_artist: known_artists = {a["tag"].lower().strip() for a in warehouse.get_all_artists()} parsed.artists = [] parsed.general_tags = [ t for t in parsed.general_tags if t.lower().strip() not in known_artists ] if strip_lora: def _is_lora(t: str) -> bool: tl = t.lower() if ":" in tl or "<" in tl or ">" in tl: return True if "lora" in tl or "loha" in tl or "lycoris" in tl or "embedding" in tl: return True return False parsed.general_tags = [t for t in parsed.general_tags if not _is_lora(t)] if strip_meta: parsed.meta_tags = [] return parsed def _apply_blacklist(tags: list[str], exclude: set[str]) -> list[str]: if not exclude: return tags return [t for t in tags if t.lower().strip() not in exclude] def _ensure_min_tags( tags: list[str], min_tags: int, warehouse: "TagWarehouse", protected: set[str], user_heads: set[str], exclude: set[str], max_rating: str, rng: "random.Random", parsed: "ParsedPrompt | None" = None, intent: str | None = None, ) -> list[str]: """Top up a tag list to at least `min_tags` (D). Candidates are drawn from the tag pools and selected through the semantic picker (intent/co-occurrence/harmony), so the fill tags are contextually relevant instead of being random pool noise. """ if min_tags <= 0: return tags out = list(tags) used = {t.lower().strip() for t in out} | exclude candidates: list[str] = [] cats = [c for c in warehouse.pools if c not in ("nsfw", "furry")] rng.shuffle(cats) for cat in cats: if len(candidates) >= min_tags * 4: break pool = warehouse.get_pool(cat) if pool is None: continue for tag in pool.get_all_tags(max_rating=max_rating): tl = tag.lower().strip() if tl in used: continue if _is_negative_quality(tag): continue if has_synonym_conflict(tag, out + candidates): continue if _head(tag) in user_heads and tl not in protected: continue candidates.append(tag) used.add(tl) if len(candidates) >= min_tags * 4: break if not candidates: return out need = min_tags - len(out) if need <= 0: return out attempts = 0 while len(out) < min_tags and candidates and attempts < min_tags * 6: attempts += 1 if parsed is None: picked = rng.sample(candidates, min(need, len(candidates))) else: picked = semantic_pick_tags(candidates, need, parsed, rng, set(), list(out), intent=intent) added_any = False for tag in picked: if len(out) >= min_tags: break new_out, was_added = _resolve_and_replace(tag, out, warehouse, protected) if was_added: out = new_out added_any = True if not added_any: # All current picks were rejected (conflicts) — drop them and retry. candidates = [c for c in candidates if c not in picked] need = min_tags - len(out) return out def _remove_intra_conflicts(tags: list[str], warehouse: TagWarehouse, protected: set[str]) -> list[str]: """Final safety pass: ensure no two tags in the result conflict. Protected (user/core) tags win; a later non-protected conflicting tag is dropped, and a later protected tag removes an earlier non-protected conflict. When BOTH tags are protected (the user's own explicit choices) we keep both rather than silently dropping one of the user's tags. Generation paths that append directly (tandems, artist signatures, fx) may bypass per-add conflict resolution, so this guarantees a clean prompt. """ from src.semantic_coherence import _scene_conflict kept: list[str] = [] for t in tags: tl = t.lower().strip() drop = False for k in list(kept): if warehouse.has_conflict(t, [k]) or _scene_conflict(tl, k.lower().strip()): kp = k.lower().strip() if tl in protected and kp in protected: continue # both are user intent: keep both if tl in protected and kp not in protected: kept.remove(k) else: drop = True break if not drop: kept.append(t) return kept def _estimate_tokens(tags: list[str]) -> int: """CLIP-style estimate: booru tag ≈ words×0.75 + 1 separator token.""" n = 0 for t in tags: words = len(t.replace("_", " ").split()) n += max(1, int(words * 0.75)) + 1 return n def _category_token_budget( resolved_categories: list[str], settings: dict, warehouse: TagWarehouse, max_rating: str, max_tokens: int, ) -> dict[str, int]: """Allocate per-category tag counts so the estimated token cost fits max_tokens. Categories with longer tags (more tokens) get fewer slots; the allocation is proportional to the category's base share so no single category can blow the budget before _balance_variation has to pop tags from the tail. """ budgets: dict[str, int] = {} if not resolved_categories or max_tokens <= 0: return budgets min_t, max_t = settings["tags_per_category"] base = (min_t + max_t) / 2.0 weights: dict[str, float] = {} for cat in resolved_categories: pool = warehouse.get_pool(cat) if pool is None: continue sample = pool.get_all_tags(max_rating=max_rating)[:200] if not sample: continue avg_cost = sum(_estimate_tokens([t]) for t in sample) / len(sample) weights[cat] = max(avg_cost, 1.0) if not weights: return budgets total_weight = sum(base * w for w in weights.values()) if total_weight <= 0: return budgets scale = max_tokens / total_weight for cat, w in weights.items(): budgets[cat] = max(min_t, min(max_t, int(base * scale))) return budgets def _tag_theme(tag: str) -> str: cats = set(get_tag_categories(tag)) for theme, cset in _THEME_GROUPS.items(): if cats & cset: return theme return "misc" def _balance_variation( tags: list[str], protected_lower: set[str], theme_budget: dict[str, int] | None, max_tokens: int, ) -> list[str]: """Trim decorative tags so no theme dominates and the token budget is respected. Protected (subject/character/series/artist/quality) tags are always kept.""" protected = [t for t in tags if t.lower().strip() in protected_lower] decorative = [t for t in tags if t.lower().strip() not in protected_lower] if theme_budget: theme_of: dict[str, str] = {} counts: dict[str, int] = {} for t in decorative: th = _tag_theme(t) theme_of[t.lower().strip()] = th counts[th] = counts.get(th, 0) + 1 for th, budget in theme_budget.items(): over = counts.get(th, 0) - budget if over <= 0: continue removed = 0 kept = [] for t in reversed(decorative): if removed < over and theme_of.get(t.lower().strip()) == th: removed += 1 continue kept.append(t) decorative = list(reversed(kept)) if max_tokens and _estimate_tokens(protected + decorative) > max_tokens: while decorative and _estimate_tokens(protected + decorative) > max_tokens: decorative.pop() return protected + decorative def _apply_fx_layer( new_general: list[str], parsed: ParsedPrompt, warehouse: TagWarehouse, rng: random.Random, fx_count: int, protected_lower: set[str], used: set[str], max_rating: str, ) -> list[str]: """Add an FX tag layer (effects/special_fx/lighting/atmosphere), co-occurrence matched to the prompt where possible. Protected tags are never displaced.""" if fx_count <= 0: return new_general fx_cats = [c for c in FX_CATEGORIES if warehouse.get_pool(c)] if not fx_cats: return new_general related: set[str] = set() if parsed.general_tags: for g in parsed.general_tags: for r in get_cooccurrence_tags(g, limit=10): rt = (r.get("tag") or "").lower().strip() if rt: related.add(rt) added = 0 attempts = 0 while added < fx_count and attempts < fx_count * 8 and fx_cats: cat = rng.choice(fx_cats) pool = warehouse.get_pool(cat) cands = pool.get_all_tags(max_rating=max_rating) if not cands: attempts += 1 continue preferred = [c for c in cands if c.lower().strip() in related] pool_choice = preferred if preferred and rng.random() < 0.7 else cands tag = rng.choice(pool_choice) tl = tag.lower().strip() if tl in used or tl in protected_lower or _is_negative_quality(tag): attempts += 1 continue new_general, ok = _resolve_and_replace(tag, new_general, warehouse, protected_lower) if ok: added += 1 used.add(tl) attempts += 1 return new_general def _score_tag_context( tag: str, parsed: ParsedPrompt, ) -> float: """Word-boundary context match between tag and user intent fields.""" score = 0.0 tag_words = set(tag.lower().replace("_", " ").split()) context_sources: list[str] = [] if parsed.subject: context_sources.append(parsed.subject.lower().strip()) if parsed.character: context_sources.append(parsed.character.lower().strip()) if parsed.series: context_sources.append(parsed.series.lower().strip()) tag_cats = get_tag_categories(tag) for ctx in context_sources: ctx_words = set(ctx.replace("_", " ").split()) if ctx_words & tag_words: score += 2.0 ctx_cats = get_tag_categories(ctx) if tag_cats and ctx_cats and (set(tag_cats) & set(ctx_cats)): score += 1.0 return min(score, SCORING["context_score_cap"]) def _cooccurrence_bonus(tag: str, context_tags: list[str]) -> float: """Sum of co-occurrence weights between tag and already-selected tags.""" if not context_tags: return 0.0 related = get_cooccurrence_tags(tag, limit=20) if not related: return 0.0 related_map = {} for r in related: rt = (r.get("tag") or "").lower().strip() if rt: related_map[rt] = r.get("weight", 1.0) bonus = 0.0 for ctx_tag in context_tags: cl = ctx_tag.lower().strip() if cl in related_map: bonus += float(related_map[cl]) return min(bonus, 5.0) def _pick_tags_weighted( candidates: list[str], count: int, parsed: ParsedPrompt, rng: random.Random, used_globals: set[str], selected_tags: list[str] | None = None, intent: str | None = None, ) -> list[str]: """Unified tag picker: delegates to semantic_pick_tags for context-aware selection.""" return semantic_pick_tags(candidates, count, parsed, rng, used_globals, selected_tags, intent=intent) def _smart_substitution( tags: list[str], chance: float, rng: random.Random, parsed: ParsedPrompt | None = None, warehouse: TagWarehouse | None = None, resolved_categories: list[str] | None = None, core_tags: set[str] | None = None, protected: set[str] | None = None, ) -> list[str]: """Context-aware substitution with cross-category fallback and scoring.""" result = list(tags) result_lower = {t.lower().strip() for t in result} # Pre-compute category pools (tag + categories) once pool_cache: dict[str, list[tuple[str, list[str]]]] = {} if warehouse: for cat in (resolved_categories or []): pool = warehouse.get_pool(cat) if pool is None: continue entries = [(pt, pt.lower().strip(), get_tag_categories(pt)) for pt in pool.get_all_tags(max_rating="explicit")] pool_cache[cat] = entries for i, tag in enumerate(result): # Protect core + user-intent tags from substitution if core_tags and tag.lower().strip() in core_tags: continue if protected and tag.lower().strip() in protected: continue if rng.random() >= chance: continue tl = tag.lower().strip() candidates = [] # 1. Synonym alternatives from same group group = _find_synonym_group(tag) substituted = False if group: for t in group: alt = t.lower().strip() if alt != tl and alt not in result_lower: candidates.append(t) if len(candidates) >= 5: rng.shuffle(candidates) best = max(candidates, key=lambda c: ( _score_tag_context(c, parsed) if parsed else 0.0 )) result = _substitute_at(result, i, best, warehouse, protected) result_lower = {t.lower().strip() for t in result} substituted = True if substituted: continue # 2. Cross-category: scan only categories the tag belongs to (not all resolved) if pool_cache: tag_cats = get_tag_categories(tag) if tag_cats: for cat in pool_cache: if cat not in tag_cats: continue for pt, ptl, pt_cats in pool_cache[cat][:30]: if ptl == tl or ptl in result_lower: continue if not (set(tag_cats) & set(pt_cats)): continue candidates.append(pt) if not candidates: continue # 3. Score and pick best other_tags = [t for j, t in enumerate(result) if j != i] best = None best_score = float("-inf") for cand in candidates: base = _score_tag_context(cand, parsed) if parsed else 0.0 cooc = _cooccurrence_bonus(cand, other_tags) penalty = 0.0 cand_syn = _find_synonym_group(cand) if cand_syn: for ot in other_tags: if ot.lower().strip() in cand_syn: penalty = 3.0 break syn_bonus = 1.0 if group and cand in group else 0.0 total = base + cooc + syn_bonus - penalty + rng.uniform(0, 0.3) if total > best_score: best_score = total best = cand if best is None: continue result = _substitute_at(result, i, best, warehouse, protected) result_lower = {t.lower().strip() for t in result} return result def _pick_random_tandem(rng: random.Random, warehouse: TagWarehouse) -> list[str]: tandems = warehouse.get_all_tandems() if not tandems: return [] tandem = rng.choice(tandems) artists = tandem.get("artists", []) tags = list(artists) for aname in artists: sig = warehouse.get_artist_signature_tags(aname) tags.extend(sig) return tags def _pick_style_tandem(rng: random.Random, warehouse: TagWarehouse, artist_style: str) -> list[str]: """Pick a tandem matching the given art style; fall back to random.""" tandems = warehouse.get_all_tandems() if not tandems: return [] if artist_style: style_artists = warehouse.get_artists_by_style(artist_style) if style_artists: style_names = {a["tag"].lower().strip() for a in style_artists} matching = [t for t in tandems if any(a.lower().strip() in style_names for a in t.get("artists", []))] if matching: tandem = rng.choice(matching) artists = tandem.get("artists", []) tags = list(artists) for aname in artists: sig = warehouse.get_artist_signature_tags(aname) tags.extend(sig) return tags return _pick_random_tandem(rng, warehouse) def _compute_tag_weights( warehouse: TagWarehouse, selected_artists: list[str] | None, ) -> dict[str, float]: all_a = warehouse.get_all_artists() max_pop = max((a.get("popularity", 0) for a in all_a), default=100) weights = {} for a in all_a: pop = a.get("popularity", 0) name = a["tag"].lower().strip() w = 1.0 + (pop / max_pop) * 0.4 weights[name] = round(min(max(w, 1.0), 1.4), 2) if selected_artists: for name in selected_artists: w = weights.get(name.lower().strip(), 1.2) weights[name.lower().strip()] = max(w, 1.2) return weights def _pick_wildcard_categories( available: list[str], count: int, parsed: ParsedPrompt, rng: random.Random, ) -> list[str]: """Pick wildcard categories contextually based on user prompt tags. Noise is low enough (0.2) that category alignment still dominates but the choice is not fully deterministic across variations. """ if not available or count <= 0: return [] cat_scores: dict[str, float] = {} for tag in parsed.general_tags: for cat in get_tag_categories(tag): cat_scores[cat] = cat_scores.get(cat, 0) + 2.0 if parsed.subject: for cat in get_tag_categories(parsed.subject): cat_scores[cat] = cat_scores.get(cat, 0) + 1.0 scored = [(cat, cat_scores.get(cat, 0.0) + rng.uniform(0, SCORING["noise_max"])) for cat in available] scored.sort(key=lambda x: -x[1]) return [c for c, _ in scored[:count]] def _adjust_settings_by_prompt_length( settings: dict, parsed: ParsedPrompt, total_avail_pools: int, ) -> dict: """Adjust creativity settings based on prompt length.""" adjusted = dict(settings) tag_count = len(parsed.general_tags or []) if tag_count <= 2: min_t, max_t = adjusted["tags_per_category"] adjusted["tags_per_category"] = (min_t + 1, max(max_t + 1, min_t + 2)) adjusted["wildcard_categories"] = min(adjusted["wildcard_categories"] + 1, total_avail_pools) adjusted["replacement_rate"] = min(adjusted["replacement_rate"] + 0.1, 0.8) elif tag_count >= 12: min_t, max_t = adjusted["tags_per_category"] adjusted["tags_per_category"] = (max(1, min_t - 1), max(1, max_t - 1)) adjusted["replacement_rate"] = max(0.0, adjusted["replacement_rate"] - 0.1) return adjusted def generate_variations( parsed: ParsedPrompt, selected_categories: list[str], num_variations: int = 5, creativity: str = "medium", model: str = "anima", rating: str = "pg", warehouse: TagWarehouse = None, artist_style: str = "", selected_artists: list[str] | None = None, use_tandems: bool = False, selected_tandem: dict | None = None, weight_mode: str = "off", mode: str = "standard", web_enrich: bool = False, selected_presets: list[str] | None = None, fx_count: int = 0, seed: int | None = None, exclude_tags: list[str] | None = None, strip_quality: bool = False, strip_artist: bool = False, strip_lora: bool = False, strip_meta: bool = False, min_tags: int = 0, output_format: str = "prompt", ) -> list[str]: if warehouse is None: warehouse = TagWarehouse() exclude = _normalize_exclude(exclude_tags) parsed.general_tags = _clean_general_tags(parsed.general_tags) parsed.general_tags = _apply_blacklist(parsed.general_tags, exclude) parsed = _apply_strip_flags( parsed, warehouse, strip_quality, strip_artist, strip_lora, strip_meta ) if mode == "rewrite": return full_rewrite( parsed=parsed, warehouse=warehouse, model=model, rating=rating, creativity=creativity, num_variations=num_variations, seed=seed, selected_categories=selected_categories, selected_presets=selected_presets, fx_count=fx_count, weight_mode=weight_mode, artist_style=artist_style, selected_artists=selected_artists, use_tandems=use_tandems, selected_tandem=selected_tandem, web_enrich=web_enrich, exclude_tags=exclude_tags, strip_quality=strip_quality, strip_artist=strip_artist, strip_lora=strip_lora, strip_meta=strip_meta, min_tags=min_tags, output_format=output_format, ) settings = CREATIVITY_SETTINGS.get(creativity, CREATIVITY_SETTINGS["medium"]) settings = _adjust_settings_by_prompt_length(settings, parsed, len(warehouse.pools)) max_rating = MAX_RATING_MAP.get(rating, "sfw") results = [] all_new_tags_per_variation: list[list[str]] = [] base_general = list(parsed.general_tags) if selected_categories: parsed = apply_synonym_filter(parsed, selected_categories, warehouse, model, rating) base_general = list(parsed.general_tags) protected_lower = _protected_set(parsed) # Preset bundle tags are user-intended additions: protect them from the # balancing/dedup passes so an applied preset is never silently dropped. for _p in (selected_presets or []): for _t in get_preset_bundle_tags(_p): protected_lower.add(_t.lower().strip()) # Head nouns the user already specified (e.g. "hair" from "blue hair"); we # avoid stacking a second same-head tag like "purple hair" on top of it. user_heads = {_head(t) for t in base_general if len(t.split()) >= 2} # Cross-category injection: add related categories via rewrite_map resolved_categories = list(selected_categories) AUTO_EXCLUDE = {"furry", "nsfw"} if settings["concept_cross"] and parsed.general_tags: for tag in parsed.general_tags: tag_cats = get_tag_categories(tag) for cat in tag_cats: if cat not in resolved_categories and warehouse.get_pool(cat) is not None: resolved_categories.append(cat) resolved_categories = [ c for c in resolved_categories if c in selected_categories or c not in AUTO_EXCLUDE ] # Wildcard categories: add context-relevant extra categories if settings["wildcard_categories"] > 0: all_avail = [c for c in warehouse.pools if c not in resolved_categories and c not in AUTO_EXCLUDE] wild_rng = random.Random(seed) if seed is not None else random.Random() extra = _pick_wildcard_categories(all_avail, settings["wildcard_categories"], parsed, wild_rng) resolved_categories.extend(extra) _skip_animal = has_human_subject(base_general) and "animal" not in selected_categories # Artist lookup map is loop-invariant — build once, reuse per variation. known_artists_map = {a["tag"].lower().strip(): a["tag"] for a in warehouse.get_all_artists()} core_tags, decorative_tags = split_core_decorative(base_general) core_set = {t.lower().strip() for t in core_tags} # Theme budget for adaptive decorative tag allocation (intent is computed ONCE # per generation call and threaded through the per-category picker). intent = detect_intent(parsed) theme_budget = compute_theme_budget(intent, resolved_categories, settings["tags_per_category"]) cat_to_theme: dict[str, str] = {} for theme, cats in _THEME_GROUPS.items(): for cat in cats: cat_to_theme[cat] = theme # Per-category token budget (loop-invariant): keeps the whole variation under # MAX_TOKENS_DEFAULT before tags are even picked, instead of trimming at the end. cat_token_budget = _category_token_budget( resolved_categories, settings, warehouse, max_rating, MAX_TOKENS_DEFAULT ) base_seed = seed for var_idx in range(num_variations): var_seed = (base_seed + var_idx * 7919) if base_seed is not None else random.randint(0, 2**31 - 1) + var_idx * 7919 rng = random.Random(var_seed) variant = deepcopy(parsed) new_general = list(base_general) new_added = [] if use_tandems and not selected_tandem and not selected_artists: tandem_tags = _pick_style_tandem(rng, warehouse, artist_style) existing_lower = {t.lower().strip() for t in new_general} for t in tandem_tags: if t.lower().strip() not in existing_lower: new_general.append(t) new_added.append(t) existing_lower.add(t.lower().strip()) if use_tandems and selected_tandem and not selected_artists: tandem_artists = selected_tandem.get("artists", []) existing_lower = {t.lower().strip() for t in new_general} for aname in tandem_artists: if aname.lower().strip() not in existing_lower: new_general.append(aname) new_added.append(aname) existing_lower.add(aname.lower().strip()) sig = warehouse.get_artist_signature_tags(aname) for st in sig: if warehouse.tag_exceeds_rating(st, max_rating): continue if st.lower().strip() not in existing_lower: new_general.append(st) new_added.append(st) existing_lower.add(st.lower().strip()) if artist_style and not selected_artists and not use_tandems and not selected_tandem: style_artists = warehouse.get_artists_by_style(artist_style) if style_artists: pool = rng.sample(style_artists, min(3, len(style_artists))) for a in pool: new_general.append(a["tag"]) new_added.append(a["tag"]) if selected_artists: for aname in selected_artists: if aname not in new_general: new_general.append(aname) new_added.append(aname) sig_tags = warehouse.get_artist_signature_tags(aname) existing_lower = {t.lower().strip() for t in new_general} for st in sig_tags: if warehouse.tag_exceeds_rating(st, max_rating): continue if st.lower().strip() not in existing_lower: new_general.append(st) new_added.append(st) existing_lower.add(st.lower().strip()) # Replacement rate: remove some existing user tags proportionally # Quality tags are weighted lower to preserve them; core tags are protected if settings["replacement_rate"] > 0 and base_general: # Never remove the user's protected intent (subject design / explicit # category tags) or core tags; only replace flexible user tags so that # variations stay true to what the user actually asked for. user_tags = [ t for t in new_general if t in base_general and t.lower().strip() not in core_set and t.lower().strip() not in protected_lower ] if user_tags: n_replace = max(1, int(len(user_tags) * settings["replacement_rate"])) quality_keywords = {"score", "masterpiece", "quality", "aesthetic", "detailed"} weights = [] for t in user_tags: tl = t.lower().strip() is_quality = any(kw in tl for kw in quality_keywords) weights.append(0.2 if is_quality else 1.0) to_remove = rng.choices(user_tags, weights=weights, k=min(n_replace, len(user_tags))) to_remove = list(dict.fromkeys(to_remove)) for t in to_remove: if t in new_general: new_general.remove(t) used_globals = {t.lower().strip() for t in new_general} # Track per-variation theme usage var_theme_usage: dict[str, int] = {} for cat in resolved_categories: if cat == "animal" and _skip_animal: continue pool = warehouse.get_pool(cat) if pool is None: continue min_t, max_t = settings["tags_per_category"] theme = cat_to_theme.get(cat, "misc") theme_max = theme_budget.get(theme, max_t * 2) used_this_theme = var_theme_usage.get(theme, 0) # Reduce count if theme budget is exceeded (and never exceed the # category's pre-allocated token budget). local_max = max(min_t, min(max_t, theme_max - used_this_theme)) local_max = min(local_max, cat_token_budget.get(cat, local_max)) count = rng.randint(min_t, local_max) if local_max >= min_t else min_t candidates = pool.get_all_tags(max_rating=max_rating) if not candidates: continue picked = _pick_tags_weighted(candidates, count, parsed, rng, used_globals, new_general, intent=intent) for tag in picked: if _is_negative_quality(tag): continue if has_synonym_conflict(tag, new_general): continue # Don't stack a second same-head tag on a tag the user already gave. if _head(tag) in user_heads and tag.lower().strip() not in protected_lower: continue new_general, was_added = _resolve_and_replace(tag, new_general, warehouse, protected_lower) if was_added: new_added.append(tag) used_globals.add(tag.lower().strip()) var_theme_usage[theme] = used_this_theme + 1 used_this_theme += 1 # Smart substitution pass: context-aware replacement with cross-category fallback if settings["substitution_chance"] > 0: new_general = _smart_substitution(new_general, settings["substitution_chance"], rng, parsed, warehouse, resolved_categories, core_set, protected_lower) # Preset overlay (protected additions) + FX tag layer (Standard Varry). for pname in (selected_presets or []): for tag in get_preset_bundle_tags(pname): new_general, _ = _resolve_and_replace(tag, new_general, warehouse, protected_lower) new_general = _apply_fx_layer(new_general, parsed, warehouse, rng, fx_count, protected_lower, used_globals, max_rating) # Web enrichment: co-occurrence + Danbooru tags keyed on the user's # subject/character (falls back to local co-occurrence data offline). if web_enrich: from src.tag_searcher import enrich_prompt_tags enrich = enrich_prompt_tags(variant, warehouse, max_tags=5, user_tags=parsed.general_tags) for t in enrich: new_general, _ = _resolve_and_replace(t, new_general, warehouse, protected_lower) variant.general_tags = new_general for prev_tags in all_new_tags_per_variation: diversity = _min_diversity_index(new_added, prev_tags) if diversity < settings["diversity_threshold"] and num_variations > 1: extra_seed = rng.randint(0, 2**31 - 1) re_rng = random.Random(extra_seed) extra_candidates = [] for cat in resolved_categories: pool = warehouse.get_pool(cat) if pool is None: continue for tag in pool.get_all_tags(max_rating=max_rating): if _is_negative_quality(tag): continue if _head(tag) in user_heads and tag.lower().strip() not in protected_lower: continue if tag.lower().strip() in used_globals: continue extra_candidates.append(tag) if extra_candidates: picked = semantic_pick_tags( extra_candidates, 2, parsed, re_rng, set(), variant.general_tags, intent=intent, ) for tag in picked: variant.general_tags, _ = _resolve_and_replace(tag, variant.general_tags, warehouse, protected_lower) break all_new_tags_per_variation.append(new_added) variant.general_tags = filter_subject_conflicts(variant.general_tags, base_general) variant.general_tags = smart_dedup(variant.general_tags, model=model) variant.general_tags = [t for t in variant.general_tags if not _is_negative_quality(t)] variant.general_tags = _remove_intra_conflicts(variant.general_tags, warehouse, protected_lower) # Move artist tags into variant.artists (loop-invariant lookup map). _extract_artists_from_general(variant, known_artists_map) # Smart balancing: cap per-theme dominance + token budget, strip metadata noise. variant.general_tags = _balance_variation(variant.general_tags, protected_lower, theme_budget, MAX_TOKENS_DEFAULT) variant.general_tags = [t for t in variant.general_tags if t.lower().strip() not in RESTRICTIVE_TAGS] variant.general_tags = _apply_blacklist(variant.general_tags, exclude) if min_tags > 0: variant.general_tags = _ensure_min_tags( variant.general_tags, min_tags, warehouse, protected_lower, user_heads, exclude, max_rating, rng, parsed=parsed, intent=intent, ) # Principled ordering: identity/composition lead, ambiance/effects trail # (SD attends to earlier tokens more, so random shuffle is harmful). variant.general_tags = _smart_order_tags(variant.general_tags) tag_weights = None if weight_mode != "off": tag_weights = _compute_tag_weights(warehouse, selected_artists) result = format_prompt( variant, model=model, rating=rating, quality_enabled=("quality" in selected_categories and not strip_quality), weight_mode=weight_mode, tag_weights=tag_weights, output_format=output_format, ) results.append(result) return results def full_rewrite( parsed: ParsedPrompt, warehouse: TagWarehouse, model: str = "anima", rating: str = "pg", creativity: str = "medium", num_variations: int = 5, seed: int | None = None, selected_categories: list[str] | None = None, selected_presets: list[str] | None = None, fx_count: int = 0, weight_mode: str = "off", artist_style: str = "", selected_artists: list[str] | None = None, use_tandems: bool = False, selected_tandem: dict | None = None, web_enrich: bool = False, exclude_tags: list[str] | None = None, strip_quality: bool = False, strip_artist: bool = False, strip_lora: bool = False, strip_meta: bool = False, min_tags: int = 0, output_format: str = "prompt", ) -> list[str]: """Full Rewrite mode: keep subject + character (the original concept), rebuild every other tag from scratch so each variation is a genuinely different but on-theme prompt. A per-variation 'angle' preset shifts mood/style/setting.""" if warehouse is None: warehouse = TagWarehouse() exclude = _normalize_exclude(exclude_tags) parsed.general_tags = _clean_general_tags(parsed.general_tags) parsed.general_tags = _apply_blacklist(parsed.general_tags, exclude) parsed = _apply_strip_flags( parsed, warehouse, strip_quality, strip_artist, strip_lora, strip_meta ) selected_presets = selected_presets or [] selected_categories = selected_categories or [] settings = CREATIVITY_SETTINGS.get(creativity, CREATIVITY_SETTINGS["medium"]) settings = _adjust_settings_by_prompt_length(settings, parsed, len(warehouse.pools)) max_rating = MAX_RATING_MAP.get(rating, "sfw") original_intent = detect_intent(parsed) protected = _protected_set(parsed, design_only=True) # Preserve user-specific names (characters/series/artists not in our pools) # so Full Rewrite keeps the identity instead of discarding it. named = { t.lower().strip() for t in parsed.general_tags if t.lower().strip() not in protected and not _is_known_tag(t) } protected |= named # Head nouns the user already specified (e.g. "hair" from "blue hair"); we # avoid stacking a second same-head tag like "neon hair" on top of it. user_heads = {_head(t) for t in parsed.general_tags if len(t.split()) >= 2} all_preset_keys = list(PRESETS.keys()) results: list[str] = [] cats = [c for c in selected_categories if c not in ("nsfw", "furry") and warehouse.get_pool(c)] if not cats: cats = [c for c in warehouse.pools if c not in ("nsfw", "furry")] cat_to_theme: dict[str, str] = {} for theme, cs in _THEME_GROUPS.items(): for c in cs: cat_to_theme[c] = theme cat_token_budget = _category_token_budget( cats, settings, warehouse, max_rating, MAX_TOKENS_DEFAULT ) base_seed = seed for var_idx in range(num_variations): var_seed = (base_seed + var_idx * 7919) if base_seed is not None else random.randint(0, 2 ** 31 - 1) + var_idx * 7919 rng = random.Random(var_seed) work = deepcopy(parsed) # Keep the user's protected design tags (hair/eyes/clothing/etc.); rebuild # everything else from scratch so the concept is preserved but fresh. preserved = [t for t in parsed.general_tags if t.lower().strip() in protected] work.general_tags = list(preserved) preserved_cats = set() for t in preserved: preserved_cats.update(get_tag_categories(t)) angle = rng.choice(all_preset_keys) presets_this = list(selected_presets) + [angle] new_general: list[str] = list(preserved) used: set[str] = {t.lower().strip() for t in preserved} theme_budget = compute_theme_budget(original_intent, cats, settings["tags_per_category"]) for cat in cats: if cat in preserved_cats: continue pool = warehouse.get_pool(cat) if pool is None: continue min_t, max_t = settings["tags_per_category"] theme = cat_to_theme.get(cat, "misc") theme_max = theme_budget.get(theme, max_t * 2) count = rng.randint(min_t, max(min_t, min(max_t, theme_max))) count = min(count, cat_token_budget.get(cat, count)) cands = pool.get_all_tags(max_rating=max_rating) if not cands: continue picked = _pick_tags_weighted(cands, count, work, rng, used, new_general, intent=original_intent) for tag in picked: if _is_negative_quality(tag): continue if has_synonym_conflict(tag, new_general): continue if _head(tag) in user_heads and tag.lower().strip() not in protected: continue new_general, _ = _resolve_and_replace(tag, new_general, warehouse, protected) used.add(tag.lower().strip()) for p in presets_this: for tag in get_preset_bundle_tags(p): new_general, _ = _resolve_and_replace(tag, new_general, warehouse, protected) new_general = _apply_fx_layer(new_general, parsed, warehouse, rng, fx_count, protected, used, max_rating) # Web enrichment: co-occurrence + Danbooru tags keyed on the user's # subject/character (falls back to local co-occurrence data offline). if web_enrich: from src.tag_searcher import enrich_prompt_tags enrich = enrich_prompt_tags(work, warehouse, max_tags=5, user_tags=parsed.general_tags) for t in enrich: new_general, _ = _resolve_and_replace(t, new_general, warehouse, protected) new_general = _balance_variation(new_general, protected, theme_budget, MAX_TOKENS_DEFAULT) new_general = [t for t in new_general if t.lower().strip() not in RESTRICTIVE_TAGS] new_general = [t for t in new_general if not _is_negative_quality(t)] new_general = _apply_blacklist(new_general, exclude) new_general = _remove_intra_conflicts(new_general, warehouse, protected) if min_tags > 0: new_general = _ensure_min_tags( new_general, min_tags, warehouse, protected, user_heads, exclude, max_rating, rng, parsed=work, intent=original_intent, ) # Principled ordering: identity/composition lead, ambiance/effects trail. new_general = _smart_order_tags(new_general) work.general_tags = new_general quality_on = "quality" in selected_categories and not strip_quality tag_weights = None if weight_mode != "off": tag_weights = _compute_tag_weights(warehouse, selected_artists) result = format_prompt( work, model=model, rating=rating, quality_enabled=quality_on, weight_mode=weight_mode, tag_weights=tag_weights, output_format=output_format, ) results.append(result) return results _NEG_TEMPLATES_PATH = _os.path.join( _os.path.dirname(_os.path.dirname(__file__)), "data", "negative_templates.json" ) def _load_negative_templates() -> dict: try: with open(_NEG_TEMPLATES_PATH, "r", encoding="utf-8") as f: return _json.load(f) except (FileNotFoundError, _json.JSONDecodeError): return {} _NEG = _load_negative_templates() NEGATIVE_PROMPTS: list[str] = _NEG.get("templates", []) _RATING_SAFETY_ADDONS: dict[str, str] = _NEG.get("rating_addons", {}) _SFWMODEL_NEGATIVE_ADDON: str = _NEG.get("sfwmodel_addon", "") _ILLUSTRIOUS_NEGATIVE_ADDON: str = _NEG.get("illustrious_addon", "") _ANIMAL_NEGATIVE_ADDON: str = _NEG.get("animal_addon", "") _HUMAN_NEGATIVE_ADDON: str = _NEG.get("human_addon", "") # Intent → tags that fight the detected scene type. These get negated so the # negative prompt opposes what will be generated (portrait → not a wide shot, # environment → not a close-up, etc.). Values are validated against the tag # pools so no phantom tags leak into the output. _INTENT_NEGATIVE_MAP = { "portrait": ["wide shot", "full body", "dutch angle"], "action": ["lying", "sitting"], "environment": ["close-up", "extreme close-up", "face focus"], "horror": ["cheerful", "bright colors", "pastel colors", "innocent"], "romantic": ["dark atmosphere", "horror"], "fantasy": ["photorealistic", "realistic"], } def _get_inverted_conflicts( tags: list[str], max_count: int = 3, rng: random.Random | None = None, ) -> list[str]: """Generate negative tags by inverting user tags' conflict/synonym groups.""" if not tags: return [] seen = set() conflicts: list[str] = [] for tag in tags: tl = tag.lower().strip() if tl in seen: continue seen.add(tl) group = _find_conflict_only_group(tag) if group and len(group) > 1: for gt in group: gtl = gt.lower().strip() if gtl != tl and gtl not in seen: conflicts.append(gt) seen.add(gtl) if len(conflicts) >= max_count * 3: break if len(conflicts) >= max_count * 3: break if rng and len(conflicts) > max_count: return rng.sample(conflicts, max_count) return conflicts[:max_count] def generate_negative_prompt( parsed: ParsedPrompt, selected_categories: list[str], num_variations: int = 5, rating: str = "pg", warehouse: TagWarehouse = None, model: str = "anima", positive_tags: list[str] | None = None, extra_negative: list[str] | None = None, output_format: str = "prompt", ) -> list[str]: results: list[str] = [] max_rating = MAX_RATING_MAP.get(rating, "sfw") has_human = bool(parsed.subject and parsed.subject not in ("no_humans", "no humans")) is_animal = "animal" in selected_categories # Intent inversion source: tags that fight the detected scene type are # negated so the negative prompt opposes what will be generated. intent = detect_intent(parsed) conflict_neg_intent: list[str] = list(_INTENT_NEGATIVE_MAP.get(intent, [])) base_pool = list(NEGATIVE_PROMPTS) # Build a conflict-inversion source from the user prompt AND a sample of # the selected category pools, so the negative prompt opposes what will be # generated (not just the literal user tags). conflict_source: list[str] = list(parsed.general_tags or []) if warehouse is not None and selected_categories: for cat in selected_categories: pool = warehouse.get_pool(cat) if pool is None: continue for t in pool.get_all_tags(max_rating=max_rating)[:5]: if t.lower().strip() not in {c.lower().strip() for c in conflict_source}: conflict_source.append(t) positive_lower = set() for pt in (positive_tags or []): for tok in str(pt).split(","): tl = tok.strip().lower() if tl: if tl.startswith("(") and tl.endswith(")") and ":" in tl: tl = tl[1:-1].rsplit(":", 1)[0].strip() positive_lower.add(tl) for var_idx in range(num_variations): seed_bytes = f"neg:{var_idx}:{rating}:{model}:{','.join(sorted(selected_categories))}".encode() seed = int.from_bytes(hashlib.sha256(seed_bytes).digest()[:4], "big") & 0x7FFFFFFF rng = random.Random(seed) primary_idx = rng.randint(0, len(base_pool) - 1) primary = base_pool[primary_idx] secondary_idx = rng.randint(0, len(base_pool) - 1) while secondary_idx == primary_idx and len(base_pool) > 1: secondary_idx = rng.randint(0, len(base_pool) - 1) secondary = base_pool[secondary_idx] parts = primary.split(", ") secondary_parts = secondary.split(", ") extra = rng.sample(secondary_parts, min(3, len(secondary_parts))) for e in extra: if e not in parts: parts.append(e) # Rating-specific safety addons safety_pool = _RATING_SAFETY_ADDONS.get(max_rating, _RATING_SAFETY_ADDONS.get("sfw", "")) if safety_pool: safety_tags = safety_pool.split(", ") parts.extend(rng.sample(safety_tags, min(2, len(safety_tags)))) if model == "anima" and rng.random() < 0.4: swf_addons = _SFWMODEL_NEGATIVE_ADDON.split(", ") parts.extend(rng.sample(swf_addons, min(2, len(swf_addons)))) if model == "illustrious" and rng.random() < 0.3: ill_addons = _ILLUSTRIOUS_NEGATIVE_ADDON.split(", ") parts.extend(rng.sample(ill_addons, min(2, len(ill_addons)))) if has_human and not is_animal: human_addons = _HUMAN_NEGATIVE_ADDON.split(", ") parts.extend(rng.sample(human_addons, min(3, len(human_addons)))) if is_animal: animal_addons = _ANIMAL_NEGATIVE_ADDON.split(", ") for a in animal_addons: if a in parts: parts.remove(a) # Conflict inversion: add 1-2 opposing tags per variation if conflict_source: conflict_neg = _get_inverted_conflicts(conflict_source, max_count=2, rng=rng) parts.extend(conflict_neg) # Intent inversion: negate tags that fight the detected scene type. if conflict_neg_intent and rng.random() < 0.6: neg_pool = [t for t in conflict_neg_intent if t.lower() not in positive_lower] if neg_pool: parts.extend(rng.sample(neg_pool, min(2, len(neg_pool)))) seen = set() deduped = [] for p in parts: pl = p.strip().lower() if not pl: continue if pl in seen: continue # Never negate a tag that is actually present in the positive prompt. if pl in positive_lower: continue seen.add(pl) deduped.append(p.strip()) # Mirror user blacklist into the negative prompt when requested (A). if extra_negative: for en in extra_negative: enl = en.strip().lower() if enl and enl not in positive_lower and enl not in seen: seen.add(enl) deduped.append(en.strip()) deduped = [normalize_tag(p, output_format) for p in deduped] results.append(", ".join(deduped)) return results