211 lines
8.4 KiB
Python
211 lines
8.4 KiB
Python
"""Per-user character alias persistence for the in-ComfyUI alias manager."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import tempfile
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
try:
|
|
import folder_paths
|
|
except ImportError:
|
|
folder_paths = None
|
|
|
|
|
|
ALIAS_FILENAME = "#character_alias_map.txt"
|
|
MAX_USER_ALIASES = 1000
|
|
MAX_USER_GROUPS = 100
|
|
_LANGUAGE_RE = re.compile(r"^[A-Za-z0-9_-]+$")
|
|
_GROUP_RE = re.compile(r"^##\s*group\s*:\s*(.+?)\s*$", re.IGNORECASE)
|
|
_MANAGED_COMMENTS = {
|
|
"Character Alias Map - managed by TTS Audio Suite",
|
|
"User aliases in this file override example and model-folder aliases.",
|
|
"Format: Alias = Character_Name, optional_language",
|
|
}
|
|
|
|
|
|
def get_user_alias_file() -> Optional[str]:
|
|
"""Return the current ComfyUI profile's highest-priority alias file."""
|
|
if folder_paths is None:
|
|
return None
|
|
base_dir = folder_paths.get_system_user_directory("tts_audio_suite")
|
|
return os.path.join(base_dir, "voices", ALIAS_FILENAME)
|
|
|
|
|
|
def parse_alias_line(line: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:
|
|
"""Parse the legacy equals or tab-separated alias formats."""
|
|
alias = target = language = None
|
|
if "=" in line:
|
|
alias, right_side = (part.strip() for part in line.split("=", 1))
|
|
if "," in right_side:
|
|
target, language = (part.strip() for part in right_side.split(",", 1))
|
|
else:
|
|
target = right_side
|
|
elif "\t" in line:
|
|
parts = [part.strip() for part in line.split("\t") if part.strip()]
|
|
if len(parts) >= 2:
|
|
alias, target = parts[:2]
|
|
language = parts[2] if len(parts) >= 3 else None
|
|
|
|
if not alias or not target:
|
|
return None, None, None
|
|
return alias, target, language or None
|
|
|
|
|
|
def parse_alias_document(path: str, infer_legacy_groups: bool = False) -> List[Dict[str, Any]]:
|
|
"""Read aliases as ordered groups while retaining ordinary comments as notes."""
|
|
groups: List[Dict[str, Any]] = [{"name": "Ungrouped", "notes": [], "aliases": [], "_explicit": False}]
|
|
current = groups[0]
|
|
with open(path, "r", encoding="utf-8") as stream:
|
|
for raw_line in stream:
|
|
line = raw_line.strip()
|
|
if not line:
|
|
continue
|
|
group_match = _GROUP_RE.match(line)
|
|
if group_match:
|
|
current = {"name": group_match.group(1).strip(), "notes": [], "aliases": [], "_explicit": True}
|
|
groups.append(current)
|
|
continue
|
|
if line.startswith("#"):
|
|
note = line.lstrip("#").strip()
|
|
if infer_legacy_groups and note.endswith(":"):
|
|
current = {"name": note[:-1].strip(), "notes": [], "aliases": [], "_explicit": False}
|
|
groups.append(current)
|
|
continue
|
|
if note and note not in _MANAGED_COMMENTS:
|
|
current["notes"].append(note)
|
|
continue
|
|
alias, target, language = parse_alias_line(line)
|
|
if alias and target:
|
|
current["aliases"].append({
|
|
"alias": alias, "target": target, "language": language or "",
|
|
})
|
|
result = []
|
|
for group in groups:
|
|
if group["aliases"] or group["notes"] or group["_explicit"]:
|
|
result.append({key: value for key, value in group.items() if key != "_explicit"})
|
|
return result
|
|
|
|
|
|
def normalize_user_aliases(entries: Any) -> List[Dict[str, str]]:
|
|
"""Validate API input while preserving the user's display casing."""
|
|
if not isinstance(entries, list):
|
|
raise ValueError("aliases must be a list")
|
|
if len(entries) > MAX_USER_ALIASES:
|
|
raise ValueError(f"aliases cannot contain more than {MAX_USER_ALIASES} entries")
|
|
|
|
normalized: List[Dict[str, str]] = []
|
|
seen = set()
|
|
for index, entry in enumerate(entries, 1):
|
|
if not isinstance(entry, dict):
|
|
raise ValueError(f"alias entry {index} must be an object")
|
|
|
|
alias = str(entry.get("alias", "")).strip()
|
|
target = str(entry.get("target", "")).strip()
|
|
language = str(entry.get("language", "")).strip()
|
|
if not alias or not target:
|
|
raise ValueError(f"alias entry {index} requires both alias and character voice")
|
|
if any(character in alias for character in "=[]\r\n\t"):
|
|
raise ValueError(f"alias '{alias}' contains a reserved character")
|
|
if any(character in target for character in ",\r\n\t"):
|
|
raise ValueError(f"character voice '{target}' contains a reserved character")
|
|
if language and not _LANGUAGE_RE.fullmatch(language):
|
|
raise ValueError(f"language '{language}' must be a language code")
|
|
|
|
key = alias.casefold()
|
|
if key in seen:
|
|
raise ValueError(f"duplicate alias '{alias}'")
|
|
seen.add(key)
|
|
normalized.append({"alias": alias, "target": target, "language": language.lower()})
|
|
|
|
return normalized
|
|
|
|
|
|
def normalize_user_groups(groups: Any) -> List[Dict[str, Any]]:
|
|
"""Validate ordered UI groups and their aliases."""
|
|
if not isinstance(groups, list):
|
|
raise ValueError("groups must be a list")
|
|
if len(groups) > MAX_USER_GROUPS:
|
|
raise ValueError(f"groups cannot contain more than {MAX_USER_GROUPS} entries")
|
|
|
|
normalized_groups = []
|
|
all_aliases = []
|
|
seen_names = set()
|
|
for index, group in enumerate(groups, 1):
|
|
if not isinstance(group, dict):
|
|
raise ValueError(f"group {index} must be an object")
|
|
name = str(group.get("name", "")).strip() or "Ungrouped"
|
|
if any(character in name for character in "\r\n"):
|
|
raise ValueError(f"group '{name}' contains a reserved character")
|
|
name_key = name.casefold()
|
|
if name_key in seen_names:
|
|
raise ValueError(f"duplicate group '{name}'")
|
|
seen_names.add(name_key)
|
|
raw_notes = [str(note) for note in group.get("notes", [])]
|
|
if any("\n" in note or "\r" in note for note in raw_notes):
|
|
raise ValueError(f"group '{name}' contains a multiline note")
|
|
notes = [note.strip().lstrip("#").strip() for note in raw_notes]
|
|
notes = [note for note in notes if note]
|
|
aliases = normalize_user_aliases(group.get("aliases", []))
|
|
all_aliases.extend(aliases)
|
|
normalized_groups.append({"name": name, "notes": notes, "aliases": aliases})
|
|
|
|
# Validate aliases across groups, not only inside each group.
|
|
normalize_user_aliases(all_aliases)
|
|
return normalized_groups
|
|
|
|
|
|
def write_user_aliases(entries: Any = None, groups: Any = None) -> str:
|
|
"""Atomically replace the per-user alias override file."""
|
|
document = normalize_user_groups(groups) if groups is not None else [
|
|
{"name": "Ungrouped", "notes": [], "aliases": normalize_user_aliases(entries)}
|
|
]
|
|
alias_file = get_user_alias_file()
|
|
if not alias_file:
|
|
raise RuntimeError("ComfyUI user storage is unavailable")
|
|
|
|
os.makedirs(os.path.dirname(alias_file), exist_ok=True)
|
|
lines = [
|
|
"# Character Alias Map - managed by TTS Audio Suite",
|
|
"# User aliases in this file override example and model-folder aliases.",
|
|
"# Format: Alias = Character_Name, optional_language",
|
|
"",
|
|
]
|
|
for group_index, group in enumerate(document):
|
|
if group_index or group["name"] != "Ungrouped":
|
|
if lines[-1]:
|
|
lines.append("")
|
|
lines.append(f"## group: {group['name']}")
|
|
for note in group["notes"]:
|
|
lines.append(f"# {note}")
|
|
for entry in group["aliases"]:
|
|
suffix = f", {entry['language']}" if entry["language"] else ""
|
|
lines.append(f"{entry['alias']} = {entry['target']}{suffix}")
|
|
content = "\n".join(lines).rstrip() + "\n"
|
|
|
|
handle, temp_path = tempfile.mkstemp(
|
|
prefix=f".{ALIAS_FILENAME}.", suffix=".tmp", dir=os.path.dirname(alias_file)
|
|
)
|
|
try:
|
|
with os.fdopen(handle, "w", encoding="utf-8", newline="\n") as stream:
|
|
stream.write(content)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temp_path, alias_file)
|
|
except Exception:
|
|
try:
|
|
os.remove(temp_path)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
return alias_file
|
|
|
|
|
|
def clear_user_aliases() -> Optional[str]:
|
|
"""Remove only UI-owned aliases; inherited files are never touched."""
|
|
alias_file = get_user_alias_file()
|
|
if alias_file and os.path.exists(alias_file):
|
|
os.remove(alias_file)
|
|
return alias_file
|