import gradio as gr import html from src.i18n import t from src.prompt_parser import parse_prompt from src.tag_warehouse import TagWarehouse, STYLE_ICONS from src.variation_engine import generate_variations, generate_negative_prompt from src.ensemble_tagger import get_ensemble_tagger from src.image_tagger import _TAGGER_DEPS_OK from src.presets import ( PRESETS, PRESET_GROUPS, get_preset_categories, get_user_preset_names, save_user_preset, delete_user_preset ) from src.prompt_analyzer import analyze_prompt, format_analysis_html from src.session_history import HistoryEntry, add_to_history, get_history from src.synonym_data import reload_synonym_groups from src.tag_format import to_booru_tag, to_prompt_tag MAX_OUTPUTS = 10 warehouse = TagWarehouse() ALL_CATEGORIES = [ "quality", "lighting", "composition", "effects", "atmosphere", "framing", "style", "colors", "background", "pose", "expression", "clothing", "special_fx", "year_meta", "nsfw", "animal", "furry", "object", "food", "vehicle", "weapon", "architecture", "demon", "angelic", "hair", "body", "accessory", "season", ] CATEGORY_ICONS = { "quality": "⭐", "lighting": "💡", "composition": "📐", "effects": "🌀", "atmosphere": "🌙", "framing": "🖼️", "style": "🎨", "colors": "🌈", "background": "🏞️", "pose": "🧍", "expression": "😊", "clothing": "👗", "special_fx": "✨", "year_meta": "📅", "nsfw": "🔞", "animal": "🐾", "furry": "🐺", "object": "🏺", "food": "🍕", "vehicle": "🚗", "weapon": "⚔️", "architecture": "🏛️", "demon": "👿", "angelic": "👼", "hair": "💇", "body": "💪", "accessory": "💍", "season": "🌸", } DEFAULT_CHECKED = {"quality", "lighting", "expression"} def _build_artist_choice(a): icon = STYLE_ICONS.get(a.get('style', ''), '') style = a.get('style', '') return f"{icon} {a['tag']} [{style}]" if style else f"{icon} {a['tag']}" artist_styles = [""] + warehouse.get_artist_styles() all_artists = warehouse.get_all_artists() all_tandems = warehouse.get_all_tandems() STYLECHOICES = [(s.title() if s else t("artist_all", "en"), s) for s in artist_styles] ARTISTCHOICES = [ (_build_artist_choice(a), a["tag"]) for a in sorted(all_artists, key=lambda a: a.get("popularity", 0), reverse=True)[:50] ] TANDEMCHOICES = [(t("tandem_placeholder", "en"), None)] + [ ( f"🤝 {tand['artists'][0]} × {tand['artists'][1]} [{tand.get('style_blend','')}] ({tand.get('compatibility',0):.0%})", i, ) for i, tand in enumerate(all_tandems) ] COPY_JS = """(text) => { if (!text || !text.trim()) return text; try { navigator.clipboard.writeText(text); } catch(e) { try { const ta = document.createElement('textarea'); ta.value = text; ta.style.position = 'fixed'; ta.style.opacity = '0'; document.body.appendChild(ta); ta.focus(); ta.select(); document.execCommand('copy'); document.body.removeChild(ta); } catch(e2) {} } const btn = document.activeElement; if (btn && btn.classList.contains('whyx-copy-btn')) { const origHTML = btn.innerHTML; btn.innerHTML = ''; btn.style.borderColor = '#34D399'; btn.style.background = 'rgba(52,211,153,0.15)'; btn.style.transform = 'scale(1.1)'; setTimeout(() => { btn.innerHTML = origHTML; btn.style.borderColor = ''; btn.style.background = ''; btn.style.transform = ''; }, 1400); } return text; }""" def on_generate(prompt, model, rating, num_variations, creativity, weight_mode, mode, lang, artist_style, selected_artists, use_tandems, selected_tandem_idx, web_enrich, seed, current_preset_state, fx_count, blacklist_text, mirror_blacklist, strip_quality, strip_artist, strip_lora, strip_meta, min_tags, user_preset_state, output_format, *checks) -> tuple[str, ...]: try: if not prompt or not prompt.strip(): return tuple([""] * (MAX_OUTPUTS * 2)) parsed = parse_prompt(prompt) if parsed is None: msg = t("error_parse", lang) return tuple([msg] + [""] * (MAX_OUTPUTS - 1) + [""] * MAX_OUTPUTS) active = [c for c, on in zip(ALL_CATEGORIES, checks) if on] if not active: active = ["quality", "lighting", "expression"] if rating in ("pg", "pg13") and "nsfw" in active: active.remove("nsfw") selected_tandem = None if use_tandems and selected_tandem_idx is not None: try: idx = int(selected_tandem_idx) if 0 <= idx < len(all_tandems): selected_tandem = all_tandems[idx] except (ValueError, TypeError): pass selected_presets = [p.strip() for p in (current_preset_state or "").split(",") if p.strip()] if user_preset_state: selected_presets.extend( p.strip() for p in str(user_preset_state).split(",") if p.strip() ) fx_count = {"off": 0, "light": 2, "rich": 4}.get(fx_count, 0) # Blacklist / Tags-to-remove (A): comma- or newline-separated. exclude_tags = [ t.strip().lower() for t in str(blacklist_text or "").replace("\n", ",").split(",") if t.strip() ] mirror = bool(mirror_blacklist) results = generate_variations( parsed=parsed, selected_categories=active, num_variations=int(num_variations), creativity=creativity, model=model.lower(), rating=rating, warehouse=warehouse, artist_style=artist_style or "", selected_artists=selected_artists or [], use_tandems=bool(use_tandems), selected_tandem=selected_tandem, weight_mode=weight_mode, mode=mode, web_enrich=web_enrich == "1", selected_presets=selected_presets, fx_count=fx_count, seed=int(seed) if seed is not None and str(seed).strip() else None, exclude_tags=exclude_tags, strip_quality=bool(strip_quality), strip_artist=bool(strip_artist), strip_lora=bool(strip_lora), strip_meta=bool(strip_meta), min_tags=int(min_tags or 0), output_format=output_format or "prompt", ) out = list(results) while len(out) < MAX_OUTPUTS: out.append("") neg_results = generate_negative_prompt( parsed=parsed, selected_categories=active, num_variations=int(num_variations), rating=rating, warehouse=warehouse, model=model.lower(), positive_tags=out[:MAX_OUTPUTS], extra_negative=exclude_tags if mirror else None, output_format=output_format or "prompt", ) neg_out = list(neg_results) while len(neg_out) < MAX_OUTPUTS: neg_out.append("") add_to_history( prompt, out[:MAX_OUTPUTS], model=model, rating=rating, num_variations=int(num_variations), creativity=creativity, weight_mode=weight_mode, ) return tuple(out[:MAX_OUTPUTS]) + tuple(neg_out[:MAX_OUTPUTS]) except Exception as exc: import traceback traceback.print_exc() err_msg = t("generation_error", lang).format(exc=html.escape(str(exc))) return tuple([err_msg] + [""] * (MAX_OUTPUTS - 1) + [""] * MAX_OUTPUTS) def _preset_status(key: str, lang: str, **kwargs) -> str: return f'
{t(key, lang).format(**kwargs) if kwargs else t(key, lang)}
' def on_user_preset_save(name: str, tags: str, lang: str): name = (name or "").strip() tags_list = [t.strip() for t in str(tags or "").replace("\n", ",").split(",") if t.strip()] if not name: return gr.update(choices=get_user_preset_names()), "", _preset_status("user_preset_need_name", lang) save_user_preset(name, tags_list) return gr.update(choices=get_user_preset_names(), value=name), "", _preset_status("user_preset_saved", lang, name=name) def on_user_preset_apply(name: str, current_state: str, lang: str): if not name: return current_state, _preset_status("user_preset_pick", lang) state = (current_state or "").strip() keys = [k for k in state.split(",") if k.strip()] if state else [] if name not in keys: keys.append(name) return ",".join(keys), _preset_status("user_preset_applied", lang, name=name) def on_user_preset_delete(name: str, lang: str): if not name: return gr.update(choices=get_user_preset_names()), _preset_status("user_preset_pick", lang) delete_user_preset(name) return gr.update(choices=get_user_preset_names(), value=None), _preset_status("user_preset_deleted", lang, name=name) def on_web_search(query: str, lang: str) -> str: from src.tag_searcher import search_tags lc = "ru" if lang == "RU" else "en" if not query or not query.strip(): return f'
{t("web_search_empty", lc)}
' try: results = search_tags(query, warehouse) parts = [] if results["local"]: parts.append(f'
🔍 {t("web_search_offline", lc)} ({len(results["local"])})
') for r in results["local"][:15]: safe_tag = html.escape(r["tag"]) safe_cat = html.escape(r["category"]) parts.append(f'{safe_tag} [{safe_cat}]') if results["web"]: parts.append(f'
🌐 {t("web_search_online", lc)} ({len(results["web"])})
') for r in results["web"][:MAX_OUTPUTS]: safe_tag = html.escape(r.get("tag", "")) count = r.get("post_count", 0) parts.append(f'{safe_tag} ({count} posts)') if not parts: return f'
{t("web_enrich_no_results", lc)}
' return '
' + "".join(parts) + '
' except Exception as exc: return f'
{t("web_search_error", lc).format(exc=html.escape(str(exc)))}
' def _format_history_html(lang: str) -> str: lc = "ru" if lang == "RU" else "en" history = get_history() lines = [] lines.append("
") entries = history.get_all()[:5] if not entries: lines.append(f"
{t('history_empty', lc)}
") else: for entry in entries: liked = entry.get_liked_results() lines.append("
") lines.append("
") lines.append(f"{t('history_prompt', lc)}: ") p = entry.prompt[:80] + "..." if len(entry.prompt) > 80 else entry.prompt lines.append(f"{html.escape(p)}
") if liked: lines.append(f"
♥ {len(liked)} {t('history_liked', lc).lower()}
") lines.append("
") favs = history.get_favorites() if favs: lines.append("
") lines.append(f"
♥ {t('favorites_label', lc)}
") for entry, idx, text in favs[:5]: t_text = text[:60] + "..." if len(text) > 60 else text lines.append(f"
{html.escape(t_text)}
") lines.append("
") else: lines.append("
") lines.append(f"
{t('history_no_likes', lc)}
") lines.append("
") lines.append("
") return "\n".join(lines) def _heart_click(idx: int, lang, *results) -> tuple: lc = "ru" if lang == "RU" else "en" history = get_history() entries = history.get_all() heart_updates = [] for i in range(MAX_OUTPUTS): if entries and i < len(entries[0].results): heart_updates.append("♥" if entries[0].is_liked(i) else "♡") else: heart_updates.append("♡") if entries and idx < len(results) and results[idx]: entries[0].toggle_like(idx) heart_updates[idx] = "♥" if entries[0].is_liked(idx) else "♡" return (gr.update(value=_format_history_html(lc)),) + tuple(gr.update(value=h) for h in heart_updates) def on_preset_change(prompt, *group_and_state): all_groups = list(group_and_state[:-1]) current_active = group_and_state[-1] active = None for vals in all_groups: if vals and isinstance(vals, list) and len(vals) > 0: active = vals[0] break if active == current_active: resets = [gr.update(value=[]) for _ in PRESET_GROUPS] return (*[c in DEFAULT_CHECKED for c in ALL_CATEGORIES], prompt, "", *resets) if active: cats = get_preset_categories(active) resets = [] for group in PRESET_GROUPS: if active in group["presets"]: resets.append(gr.update(value=[active])) else: resets.append(gr.update(value=[])) # Presets are now layered inside the engine (protected additions), # so the prompt text itself is left untouched. return (*[c in cats for c in ALL_CATEGORIES], prompt, active, *resets) else: resets = [gr.update(value=[]) for _ in PRESET_GROUPS] return (*[c in DEFAULT_CHECKED for c in ALL_CATEGORIES], prompt, "", *resets) def _clear_preset_state() -> str: return "" def on_rating_change(rating: str, *checks) -> list[bool]: checks = list(checks) if rating in ("pg", "pg13"): idx = ALL_CATEGORIES.index("nsfw") if idx < len(checks): checks[idx] = False return (*checks,) def on_artist_filter_change(style: str, query: str): q = (query or "").lower().strip() filtered = all_artists if style: filtered = [a for a in filtered if a.get("style") == style] if q: filtered = [ a for a in filtered if q in a["tag"].lower() or q in a.get("style", "").lower() or any(q in t.lower() for t in a.get("signature_tags", [])) ] # Lazy-load: limit to top 50 by popularity unless searching if not q and len(filtered) > 50: filtered = sorted(filtered, key=lambda a: a.get("popularity", 0), reverse=True)[:50] choices = [(_build_artist_choice(a), a["tag"]) for a in filtered] return gr.update(choices=choices, value=[]) def _format_artist_info(artist_names: list[str], lang: str = "en") -> str: title = t("artist_preview_title", lang) if not artist_names: empty = t("artist_preview_empty", lang) return f"""
{title}
{empty}
""" lines = [] for name in artist_names: safe_name = html.escape(name) desc = warehouse.get_artist_description(name, lang) url = warehouse.get_artist_danbooru_url(name) if desc: safe_url = html.escape(url) if url else "" safe_desc = html.escape(desc) link = f"{t('danbooru_link', lang)}" if url else "" lines.append(f"""
{safe_name} {link}
{safe_desc}
""") # Similar artists from the first selected one if len(artist_names) == 1: similar = warehouse.get_similar_artists(artist_names[0], limit=3) if similar: names = ", ".join(html.escape(a["tag"]) for a in similar) label = t("similar_artists", lang) lines.append(f"
✨ {label}: {names}
") return "
" + title + "
" + "".join(lines) + "
" def on_artist_selection_change(selected_artists, lang): return gr.update(value=_format_artist_info(selected_artists or [], lang)) def on_prompt_analyze(prompt: str, lang: str): lc = "ru" if lang == "RU" else "en" data = analyze_prompt(prompt, warehouse) return gr.update(value=format_analysis_html(data, lc)) def _reload_data(lang): warehouse.reload_pools() warehouse.reload_artists() reload_synonym_groups() from src.prompt_rewriter import reload_map from src.tag_searcher import reload_cooccurrence from src.dedup_engine import reload_groups reload_map() reload_cooccurrence() reload_groups() def on_artist_style_recommendations(style: str, lang: str): title = t("artist_preview_title", lang) if not style: empty = t("artist_preview_empty", lang) return gr.update(value=f"""
{title}
{empty}
""") artists = warehouse.get_artists_by_style(style) artists = sorted(artists, key=lambda a: a.get("popularity", 0), reverse=True)[:3] if not artists: empty = t("artist_preview_empty", lang) return gr.update(value=f"""
{title}
{empty}
""") names = ", ".join(html.escape(a["tag"]) for a in artists) label = t("top_artists_in_style", lang) return gr.update(value=f"
{title}
✨ {label}: {names}
") def _build_tagger_html(result: dict, lc: str) -> str: lines = [] lines.append("
") # Natural-language caption (Qwen-VL) — shown first when the ensemble emits it. nl_caption = result.get("nl_caption") if nl_caption: lines.append( "
🗣️ " + t("tagger_caption_label", lc) + "
" f"
{html.escape(nl_caption)}
" ) # Pose summary from the pose tagger, when available. pose_tags = result.get("pose_tags") or [] ppl = result.get("people_count", 0) if pose_tags or ppl: safe = [html.escape(p) for p in pose_tags] line = f"
{t('tagger_pose_label', lc)}
" label = t("tagger_pose_single" if ppl == 1 else "tagger_pose_many", lc) if ppl: line += f"
{label.format(n=ppl)}
" if safe: line += f"
{', '.join(safe)}
" lines.append(line) lines.append(f"
{t('tagger_ratings', lc)}
") rating_colors = {"general": "#34D399", "sensitive": "#FBBF24", "questionable": "#F97316", "explicit": "#EF4444"} for name, score in result["ratings"].items(): color = rating_colors.get(name, "var(--text-dim)") pct = int(score * 100) bar = "▰" * (pct // 10) + "▱" * (10 - pct // 10) lines.append(f"
{name} {bar} ({score:.1%})
") if result["characters"]: lines.append(f"
{t('tagger_characters', lc)}
") lines.append("
") for name, score in list(result["characters"].items())[:6]: safe = html.escape(name.replace("_", " ").replace("(", "\\(").replace(")", "\\)")) lines.append(f"{safe} {score:.0%}") lines.append("
") lines.append(f"
{t('tagger_top_general', lc)}
") lines.append("
") for name, score in list(result["general"].items())[:14]: safe = html.escape(name.replace("_", " ").replace("(", "\\(").replace(")", "\\)")) pct = int(score * 100) bar = "▰" * (pct // 10) + "▱" * (10 - pct // 10) lines.append(f"
{bar}{safe}{score:.0%}
") lines.append("
") lines.append("
") return "\n".join(lines) def on_tag_image(image, gen_threshold, char_threshold, fmt, lang, progress=gr.Progress()): lc = "ru" if lang == "RU" else "en" empty_chips = gr.update(choices=[], value=[], visible=False) if image is None: return ( gr.update(value=f'
{t("tagger_no_image", lc)}
'), gr.update(visible=False), empty_chips, empty_chips, gr.update(visible=False), gr.update(visible=False), "", "", ) try: progress(0, desc=t("tagger_processing", lc)) from src.ensemble_tagger import get_ensemble_tagger ens = get_ensemble_tagger() if not _TAGGER_DEPS_OK: msg = f'
{t("tagger_unavailable", lc)}
' return ( gr.update(value=msg), gr.update(visible=False), empty_chips, empty_chips, gr.update(visible=False), gr.update(visible=False), "", "", ) progress(0.35, desc=t("tagger_ratings_label", lc)) result = ens.tag_image(image, gen_threshold, char_threshold, mode="ensemble") progress(0.9, desc=t("tagger_results", lc)) if not result["general"] and not result["characters"]: html_out = f'
{t("tagger_no_tags", lc)}
' return ( gr.update(value=html_out), gr.update(visible=False), empty_chips, empty_chips, gr.update(visible=False), gr.update(visible=False), "", "", ) html_out = _build_tagger_html(result, lc) general_names = list(result["general"].keys())[:60] char_names = list(result["characters"].keys()) general_choices = [(n.replace("_", " "), n) for n in general_names] char_choices = [(n.replace("_", " "), n) for n in char_names] raw_caption = result["caption"] esc_taglist = result["taglist"] tags_box = esc_taglist if fmt == "prompt" else raw_caption return ( gr.update(value=html_out), gr.update(visible=True), gr.update(choices=general_choices, value=general_names, visible=True), gr.update(choices=char_choices, value=char_names, visible=True), gr.update(value=tags_box, visible=True), gr.update(visible=True), raw_caption, esc_taglist, ) except Exception as exc: import traceback traceback.print_exc() err = t("tagger_error", lc).format(exc=html.escape(str(exc))) return ( gr.update(value=f'
{err}
'), gr.update(visible=False), empty_chips, empty_chips, gr.update(visible=False), gr.update(visible=False), "", "", ) def _join_selected(chips, chars, fmt): selected = (list(chips or []) + list(chars or [])) if fmt == "raw": return ", ".join(to_booru_tag(t) for t in selected) return ", ".join(to_prompt_tag(t) for t in selected) def on_tagger_selection_change(chips, chars, fmt, raw_state, esc_state): return _join_selected(chips, chars, fmt) def on_tagger_format_change(fmt, chips, chars): return _join_selected(chips, chars, fmt) def on_tagger_apply_tags(current_prompt, tags): if tags and isinstance(tags, str) and tags.strip(): if current_prompt and current_prompt.strip(): return f"{current_prompt.strip()}, {tags.strip()}" return tags.strip() return current_prompt if current_prompt else "" __all__ = [ "MAX_OUTPUTS", "ALL_CATEGORIES", "CATEGORY_ICONS", "DEFAULT_CHECKED", "warehouse", "artist_styles", "all_artists", "all_tandems", "STYLECHOICES", "ARTISTCHOICES", "TANDEMCHOICES", "COPY_JS", "on_generate", "_preset_status", "on_user_preset_save", "on_user_preset_apply", "on_user_preset_delete", "on_web_search", "_format_history_html", "_heart_click", "on_preset_change", "_clear_preset_state", "on_rating_change", "on_artist_filter_change", "_format_artist_info", "on_artist_selection_change", "on_prompt_analyze", "_reload_data", "on_artist_style_recommendations", "_build_tagger_html", "on_tag_image", "_join_selected", "on_tagger_selection_change", "on_tagger_format_change", "on_tagger_apply_tags", ]