Upload 2 files

salia_get_bam.py

@@ -0,0 +1,66 @@
+import urllib.error
+import urllib.request
+
+BAM_URL = "https://huggingface.co/saliacoel/chars/raw/main/BAM.txt"
+
+
+class GetBAM:
+    CATEGORY = "utils/text"
+    FUNCTION = "get_line"
+    RETURN_TYPES = ("STRING",)
+    RETURN_NAMES = ("line",)
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "ID": (
+                    "INT",
+                    {
+                        "default": 1,
+                        "min": 1,
+                        "max": 99999,
+                        "step": 1,
+                    },
+                ),
+            }
+        }
+
+    @classmethod
+    def IS_CHANGED(cls, ID):
+        # This node reads a remote text file that may change outside ComfyUI,
+        # so force re-execution instead of relying only on widget changes.
+        return float("nan")
+
+    def get_line(self, ID):
+        request = urllib.request.Request(
+            BAM_URL,
+            headers={"User-Agent": "ComfyUI-get_BAM/1.0"},
+        )
+
+        try:
+            with urllib.request.urlopen(request, timeout=10) as response:
+                text = response.read().decode("utf-8-sig", errors="replace")
+        except urllib.error.HTTPError as e:
+            return (f"ERROR: HTTP {e.code} while reading BAM.txt",)
+        except urllib.error.URLError as e:
+            reason = getattr(e, "reason", e)
+            return (f"ERROR: Could not reach BAM.txt ({reason})",)
+        except Exception as e:
+            return (f"ERROR: {e}",)
+
+        lines = text.splitlines()
+
+        if ID < 1 or ID > len(lines):
+            return (f"ERROR: line {ID} is out of range (1-{len(lines)})",)
+
+        return (lines[ID - 1],)
+
+
+NODE_CLASS_MAPPINGS = {
+    "get_BAM": GetBAM,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "get_BAM": "get_BAM",
+}

salia_zip_to_chars_bam.py

@@ -0,0 +1,411 @@
+import io
+import os
+import re
+import zipfile
+from typing import Dict, List, Tuple
+from urllib.parse import quote, urlparse
+from urllib.request import Request, urlopen
+
+TARGET_REPO_ID = "saliacoel/chars"
+TARGET_REPO_TYPE = "model"
+TARGET_BRANCH = "main"
+TARGET_BAM_PATH = "BAM.txt"
+CATEGORY = "Salia/HuggingFace"
+
+_INVALID_PATH_CHARS = re.compile(r'[<>:"\\|?*\x00-\x1F]')
+_NUMBERED_BAM_LINE_RE = re.compile(r"^(\s*)(\d+)\.\s+(.*)$")
+
+
+def _require_non_empty_string(value, name: str) -> str:
+    if value is None:
+        raise ValueError(f"{name} is required.")
+    value = str(value).strip()
+    if not value:
+        raise ValueError(f"{name} must not be empty.")
+    return value
+
+
+def _optional_string(value) -> str:
+    if value is None:
+        return ""
+    return str(value).strip()
+
+
+def _resolve_hf_token(hf_token: str) -> str:
+    token = _optional_string(hf_token)
+    if token:
+        return token
+
+    for env_name in ("HF_TOKEN", "HUGGINGFACE_TOKEN", "HUGGING_FACE_HUB_TOKEN"):
+        env_value = os.environ.get(env_name, "").strip()
+        if env_value:
+            return env_value
+
+    raise ValueError(
+        "A Hugging Face write token is required. Provide hf_token in the node, or set HF_TOKEN / HUGGINGFACE_TOKEN / HUGGING_FACE_HUB_TOKEN in the environment."
+    )
+
+
+def _sanitize_path_component(component: str, label: str = "path component") -> str:
+    component = _require_non_empty_string(component, label)
+    component = component.replace("\\", "/")
+    component = component.split("/")[-1]
+    component = _INVALID_PATH_CHARS.sub("_", component)
+    component = component.strip().strip(".")
+    if component in {"", ".", ".."}:
+        raise ValueError(f"Invalid {label}: {component!r}")
+    return component
+
+
+
+def _sanitize_repo_path(path_in_repo: str) -> str:
+    path_in_repo = _require_non_empty_string(path_in_repo, "path_in_repo")
+    raw_parts = path_in_repo.replace("\\", "/").split("/")
+    cleaned_parts = []
+    for part in raw_parts:
+        part = part.strip()
+        if not part or part == ".":
+            continue
+        if part == "..":
+            raise ValueError("path_in_repo must not contain '..'.")
+        cleaned_parts.append(_sanitize_path_component(part))
+    if not cleaned_parts:
+        raise ValueError("path_in_repo became empty after sanitization.")
+    return "/".join(cleaned_parts)
+
+
+
+def _quote_repo_path(path_in_repo: str) -> str:
+    return "/".join(quote(part, safe="") for part in path_in_repo.split("/"))
+
+
+
+def _repo_file_url(path_in_repo: str) -> str:
+    path_in_repo = _sanitize_repo_path(path_in_repo)
+    quoted_path = _quote_repo_path(path_in_repo)
+    return f"https://huggingface.co/{TARGET_REPO_ID}/resolve/{TARGET_BRANCH}/{quoted_path}"
+
+
+
+def _import_hf_bits():
+    try:
+        from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download
+    except Exception as exc:
+        raise RuntimeError(
+            "Missing dependency 'huggingface_hub'. Install it in the ComfyUI Python environment with: pip install huggingface_hub"
+        ) from exc
+    return HfApi, hf_hub_download, CommitOperationAdd
+
+
+
+def _download_url_bytes(url: str, hf_token: str) -> bytes:
+    url = _require_non_empty_string(url, "zip_url")
+    parsed = urlparse(url)
+    if parsed.scheme not in {"http", "https"}:
+        raise ValueError("zip_url must start with http:// or https://")
+
+    headers = {"User-Agent": "ComfyUI-Salia-Import-TMP-Zip-To-Chars/1.0"}
+    host = (parsed.netloc or "").lower()
+    token = _optional_string(hf_token)
+    if token and (host.endswith("huggingface.co") or host.endswith("hf.co")):
+        headers["Authorization"] = f"Bearer {token}"
+
+    request = Request(url, headers=headers, method="GET")
+    try:
+        with urlopen(request, timeout=90) as response:
+            return response.read()
+    except Exception as exc:
+        raise RuntimeError(f"Failed to download zip_url: {url}") from exc
+
+
+
+def _decode_text(raw: bytes) -> str:
+    for encoding in ("utf-8-sig", "utf-8", "latin-1"):
+        try:
+            return raw.decode(encoding)
+        except Exception:
+            continue
+    return raw.decode("utf-8", errors="replace")
+
+
+
+def _normalize_newlines(text: str) -> str:
+    return str(text).replace("\r\n", "\n").replace("\r", "\n")
+
+
+
+def _load_zip_character_files(zip_bytes: bytes) -> Dict[str, Dict[str, str]]:
+    character_files: Dict[str, Dict[str, str]] = {}
+
+    try:
+        zip_buffer = io.BytesIO(zip_bytes)
+        with zipfile.ZipFile(zip_buffer, "r") as zf:
+            for info in zf.infolist():
+                if info.is_dir():
+                    continue
+
+                internal_path = info.filename.replace("\\", "/").strip("/")
+                if not internal_path or internal_path.startswith("__MACOSX/"):
+                    continue
+
+                parts = [part for part in internal_path.split("/") if part]
+                if len(parts) < 2:
+                    continue
+
+                character_name = parts[0]
+                relative_path = "/".join(parts[1:])
+                if not relative_path.lower().endswith(".txt"):
+                    continue
+
+                text = _normalize_newlines(_decode_text(zf.read(info)))
+                character_bucket = character_files.setdefault(character_name, {})
+                character_bucket[relative_path] = text
+    except zipfile.BadZipFile as exc:
+        raise ValueError("zip_url did not contain a valid .zip archive.") from exc
+
+    if not character_files:
+        raise ValueError("No character folders with .txt files were found in the zip archive.")
+
+    return character_files
+
+
+
+def _extract_bam_updates_and_repo_files(character_files: Dict[str, Dict[str, str]]):
+    bam_updates: Dict[str, str] = {}
+    repo_text_files: List[Tuple[str, bytes]] = []
+
+    for character_name, files in character_files.items():
+        clean_character_name = _sanitize_path_component(character_name, "character folder name")
+
+        for relative_path, text in files.items():
+            basename = os.path.basename(relative_path).lower()
+            if basename == "bam.txt":
+                bam_line = text.strip()
+                if bam_line:
+                    bam_updates[character_name] = bam_line
+                continue
+
+            repo_relative_path = _sanitize_repo_path(f"{clean_character_name}/{relative_path}")
+            repo_text_files.append((repo_relative_path, text.encode("utf-8")))
+
+    return bam_updates, repo_text_files
+
+
+
+def _split_bam_line(line: str):
+    match = _NUMBERED_BAM_LINE_RE.match(line.rstrip("\n\r"))
+    if not match:
+        return None
+
+    indent, number, rest = match.groups()
+    rest = rest.rstrip()
+    if "," in rest:
+        character_name, suffix_rest = rest.split(",", 1)
+        suffix = "," + suffix_rest
+    else:
+        character_name = rest
+        suffix = ""
+
+    return {
+        "indent": indent,
+        "number": int(number),
+        "character_name": character_name.strip(),
+        "suffix": suffix,
+    }
+
+
+
+def _append_tag_to_existing_line(line: str, bam_tag: str):
+    stripped = line.rstrip()
+    if bam_tag in stripped:
+        return stripped, False, True
+
+    if stripped.endswith(","):
+        return stripped + " " + bam_tag, True, False
+
+    return stripped + ", " + bam_tag, True, False
+
+
+
+def _apply_bam_updates(current_bam_text: str, bam_updates: Dict[str, str]):
+    lines = _normalize_newlines(current_bam_text).split("\n") if current_bam_text else []
+
+    if lines and lines[-1] == "":
+        lines = lines[:-1]
+
+    highest_number = 0
+    for line in lines:
+        parsed = _split_bam_line(line)
+        if parsed:
+            highest_number = max(highest_number, parsed["number"])
+
+    updated_existing = 0
+    appended_new_lines = 0
+    duplicate_skips = 0
+
+    for character_name, bam_tag in bam_updates.items():
+        found = False
+        for index, line in enumerate(lines):
+            parsed = _split_bam_line(line)
+            if not parsed:
+                continue
+            if parsed["character_name"].casefold() != character_name.strip().casefold():
+                continue
+
+            new_line, changed, duplicate = _append_tag_to_existing_line(line, bam_tag)
+            lines[index] = new_line
+            if duplicate:
+                duplicate_skips += 1
+            elif changed:
+                updated_existing += 1
+            found = True
+            break
+
+        if not found:
+            highest_number += 1
+            lines.append(f"{highest_number}. {character_name.strip()}, {bam_tag}")
+            appended_new_lines += 1
+
+    if not lines:
+        return "", updated_existing, appended_new_lines, duplicate_skips
+
+    return "\n".join(lines) + "\n", updated_existing, appended_new_lines, duplicate_skips
+
+
+
+def _download_current_bam_text(hf_token: str) -> str:
+    _, hf_hub_download, _ = _import_hf_bits()
+
+    try:
+        local_path = hf_hub_download(
+            repo_id=TARGET_REPO_ID,
+            repo_type=TARGET_REPO_TYPE,
+            filename=TARGET_BAM_PATH,
+            revision=TARGET_BRANCH,
+            token=hf_token,
+        )
+        with open(local_path, "rb") as handle:
+            return _normalize_newlines(_decode_text(handle.read()))
+    except Exception as exc:
+        message = str(exc).lower()
+        if "404" in message or "not found" in message or "entry not found" in message:
+            return ""
+        raise RuntimeError(f"Failed to download current {TARGET_BAM_PATH} from {TARGET_REPO_ID}.") from exc
+
+
+
+def _commit_files(named_bytes: List[Tuple[str, bytes]], hf_token: str, commit_message: str):
+    HfApi, _, CommitOperationAdd = _import_hf_bits()
+    api = HfApi(token=hf_token)
+
+    operations = [
+        CommitOperationAdd(path_in_repo=_sanitize_repo_path(path_in_repo), path_or_fileobj=file_bytes)
+        for path_in_repo, file_bytes in named_bytes
+    ]
+
+    commit_info = api.create_commit(
+        repo_id=TARGET_REPO_ID,
+        repo_type=TARGET_REPO_TYPE,
+        revision=TARGET_BRANCH,
+        operations=operations,
+        commit_message=commit_message,
+        token=hf_token,
+    )
+
+    commit_url = getattr(commit_info, "commit_url", None)
+    if not commit_url:
+        commit_url = str(commit_info)
+    return commit_url
+
+
+class Salia_Import_TMP_zip_to_chars:
+    CATEGORY = CATEGORY
+    OUTPUT_NODE = True
+    RETURN_TYPES = ("STRING", "INT", "INT", "STRING", "STRING")
+    RETURN_NAMES = ("summary", "character_count", "uploaded_text_file_count", "bam_url", "commit_url")
+    FUNCTION = "apply_zip"
+    DESCRIPTION = (
+        "Download a zip of character folders from a URL, merge each folder's bam.txt into saliacoel/chars:BAM.txt, "
+        "and upload the other .txt files into matching character folders in saliacoel/chars."
+    )
+    SEARCH_ALIASES = [
+        "reverse tmp zip import",
+        "download zip to chars repo",
+        "merge bam txt from zip",
+    ]
+
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "zip_url": (
+                    "STRING",
+                    {
+                        "default": "https://huggingface.co/saliacoel/tmp/resolve/main/zipped_textfiles.zip",
+                        "multiline": False,
+                    },
+                ),
+                "hf_token": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "multiline": False,
+                        "placeholder": "hf_... (leave blank to use HF_TOKEN env var)",
+                    },
+                ),
+            }
+        }
+
+    def apply_zip(self, zip_url, hf_token):
+        token = _resolve_hf_token(hf_token)
+        zip_bytes = _download_url_bytes(zip_url, token)
+        character_files = _load_zip_character_files(zip_bytes)
+        bam_updates, repo_text_files = _extract_bam_updates_and_repo_files(character_files)
+
+        if not bam_updates and not repo_text_files:
+            raise ValueError("No usable .txt files were found in the zip archive.")
+
+        current_bam_text = _download_current_bam_text(token)
+        new_bam_text, updated_existing, appended_new_lines, duplicate_skips = _apply_bam_updates(
+            current_bam_text,
+            bam_updates,
+        )
+
+        commit_files = list(repo_text_files)
+        commit_files.append((TARGET_BAM_PATH, new_bam_text.encode("utf-8")))
+
+        character_count = len(character_files)
+        uploaded_text_file_count = len(repo_text_files)
+        commit_message = (
+            f"ComfyUI import {character_count} character folders from zip "
+            f"({uploaded_text_file_count} text files + {TARGET_BAM_PATH})"
+        )
+        commit_url = _commit_files(commit_files, token, commit_message)
+
+        summary = (
+            f"Processed {character_count} character folders. "
+            f"Uploaded {uploaded_text_file_count} non-bam text files. "
+            f"BAM updates: {len(bam_updates)} tags processed, "
+            f"{updated_existing} existing lines extended, "
+            f"{appended_new_lines} new lines appended, "
+            f"{duplicate_skips} duplicate tags skipped."
+        )
+
+        return (
+            summary,
+            int(character_count),
+            int(uploaded_text_file_count),
+            _repo_file_url(TARGET_BAM_PATH),
+            commit_url,
+        )
+
+
+NODE_CLASS_MAPPINGS = {
+    "Salia_Import_TMP_zip_to_chars": Salia_Import_TMP_zip_to_chars,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "Salia_Import_TMP_zip_to_chars": "Salia_Import_TMP_zip_to_chars",
+}
+
+__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]