feat: load NovelFull chapters directly for audiobooks

nodes/__init__.py

@@ -1,9 +1,10 @@
 from .loader import OmniVoiceModelLoader
 from .generator import OmniVoiceGenerate
 from .epub_loader import OmniVoiceEpubLoader
+from .web_novel_loader import OmniVoiceWebNovelLoader
 from .voice_presets import OmniVoiceVoicePreset
 from .mix_voices import OmniVoiceMixVoices
 from .voice_design import OmniVoiceVoiceDesign
 from .multi_speaker import OmniVoiceSpeaker, OmniVoiceSpeakers
 
-__all__ = ["OmniVoiceModelLoader", "OmniVoiceGenerate", "OmniVoiceEpubLoader", "OmniVoiceVoicePreset", "OmniVoiceMixVoices", "OmniVoiceVoiceDesign", "OmniVoiceSpeaker", "OmniVoiceSpeakers"]
+__all__ = ["OmniVoiceModelLoader", "OmniVoiceGenerate", "OmniVoiceEpubLoader", "OmniVoiceWebNovelLoader", "OmniVoiceVoicePreset", "OmniVoiceMixVoices", "OmniVoiceVoiceDesign", "OmniVoiceSpeaker", "OmniVoiceSpeakers"]

nodes/web_novel_loader.py

@@ -0,0 +1,239 @@
+"""Load NovelFull chapters into the same text pipeline as the EPUB loader."""
+
+import hashlib
+import json
+import logging
+import os
+from pathlib import Path
+import re
+import tempfile
+import time
+from urllib.error import HTTPError, URLError
+from urllib.parse import parse_qs, urljoin, urlsplit, urlunsplit
+from urllib.request import Request, urlopen
+
+from bs4 import BeautifulSoup
+
+try:
+    import folder_paths
+    _CACHE_DIR = Path(folder_paths.models_dir) / "omnivoice" / "web_novels"
+except ImportError:
+    _CACHE_DIR = Path.home() / ".cache" / "omnivoice" / "web_novels"
+
+_LOG = logging.getLogger(__name__)
+_CHAPTER_HEADING = re.compile(r"^chapter\s*(\d+)\b", re.IGNORECASE)
+# Match the site's injected footer, including its Cyrillic/Unicode spelling.
+_WATERMARK = re.compile(
+    r"(?:fr[eеёéè]{2}\s*)?read\s*novel\s*full\s*\.\s*(?:com|net)",
+    re.IGNORECASE,
+)
+_CREDITS = re.compile(r"^(?:translator|editor|translation)\s*:", re.IGNORECASE)
+
+
+def _validate_url(url):
+    parts = urlsplit(url.strip())
+    if (parts.scheme not in {"http", "https"}
+            or parts.hostname not in {"novelfull.net", "www.novelfull.net"}
+            or parts.username or parts.password or parts.port not in {None, 80, 443}):
+        raise ValueError("Use a public NovelFull URL on https://novelfull.net/.")
+    return parts
+
+
+def _novel_url(url):
+    parts = _validate_url(url)
+    if not re.fullmatch(r"/[^/]+\.html", parts.path):
+        raise ValueError("Paste the novel's chapter-list URL, not an individual chapter URL.")
+    # Always begin with the first page, even if the pasted URL has ?page=N.
+    return urlunsplit(("https", "novelfull.net", parts.path, "", ""))
+
+
+def _fetch_html(url):
+    _validate_url(url)
+    for attempt in range(3):
+        try:
+            request = Request(url, headers={"User-Agent": "ComfyUI-OmniVoice/1.0"})
+            with urlopen(request, timeout=20) as response:
+                _validate_url(response.geturl())
+                return response.read().decode("utf-8", errors="replace")
+        except HTTPError as exc:
+            if exc.code in {429, 500, 502, 503, 504} and attempt < 2:
+                time.sleep(2 ** (attempt + 1))
+                continue
+            raise RuntimeError(f"NovelFull returned HTTP {exc.code} for {url}.") from exc
+        except (URLError, TimeoutError, OSError) as exc:
+            if attempt < 2:
+                time.sleep(2 ** (attempt + 1))
+                continue
+            raise RuntimeError(f"Could not download {url}: {exc}") from exc
+
+
+def _read_cache(path):
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+
+
+def _write_cache(path, value):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent,
+                                     suffix=".tmp", delete=False) as handle:
+        temp_path = Path(handle.name)
+        try:
+            json.dump(value, handle, ensure_ascii=False)
+        except BaseException:
+            temp_path.unlink(missing_ok=True)
+            raise
+    try:
+        os.replace(temp_path, path)
+    finally:
+        temp_path.unlink(missing_ok=True)
+
+
+def _parse_index(html, novel_url):
+    soup = BeautifulSoup(html, "html.parser")
+    container = soup.select_one("#list-chapter, .list-chapter")
+    if container is None:
+        raise ValueError("NovelFull's chapter list was not found. The page may be blocked or its layout changed.")
+    chapter_prefix = urlsplit(novel_url).path.removesuffix(".html") + "/"
+    chapters = []
+    seen = set()
+    for anchor in container.select("a[href]"):
+        url = urljoin(novel_url, anchor["href"])
+        parts = urlsplit(url)
+        if (parts.hostname != "novelfull.net" or not parts.path.startswith(chapter_prefix)
+                or not parts.path.endswith(".html")):
+            continue
+        url = urlunsplit(("https", "novelfull.net", parts.path, "", ""))
+        title = " ".join(anchor.get_text(" ", strip=True).split())
+        if title and url not in seen:
+            seen.add(url)
+            chapters.append({"title": title, "url": url})
+    if not chapters:
+        raise ValueError("NovelFull's chapter list is empty; no audio will be generated.")
+
+    # New layout publishes the total; old layouts expose it through pagination.
+    pages = int(container.get("data-total-page", "1"))
+    for anchor in container.select("a[href]"):
+        parts = urlsplit(urljoin(novel_url, anchor["href"]))
+        if parts.hostname == "novelfull.net" and parts.path == urlsplit(novel_url).path:
+            number = parse_qs(parts.query).get("page", ["1"])[0]
+            if number.isdigit():
+                pages = max(pages, int(number))
+    if not 1 <= pages <= 1000:
+        raise ValueError("NovelFull reported an invalid number of chapter-list pages.")
+    return chapters, pages
+
+
+def _load_index(novel_url, cache_dir, refresh):
+    path = cache_dir / "index.json"
+    cached = _read_cache(path) if not refresh else None
+    if isinstance(cached, list) and cached and all(
+            isinstance(ch, dict) and ch.get("title") and ch.get("url") for ch in cached):
+        return cached
+
+    chapters, pages = _parse_index(_fetch_html(novel_url), novel_url)
+    seen = {ch["url"] for ch in chapters}
+    for page in range(2, pages + 1):
+        _LOG.info("OmniVoice: reading NovelFull chapter-list page %s/%s", page, pages)
+        time.sleep(0.25)
+        batch, _ = _parse_index(_fetch_html(f"{novel_url}?page={page}"), novel_url)
+        if not any(ch["url"] not in seen for ch in batch):
+            raise ValueError(f"NovelFull repeated chapter-list page {page}; the index is incomplete.")
+        for chapter in batch:
+            if chapter["url"] not in seen:
+                chapters.append(chapter)
+                seen.add(chapter["url"])
+    _write_cache(path, chapters)
+    return chapters
+
+
+def _parse_chapter(html, chapter):
+    soup = BeautifulSoup(html, "html.parser")
+    content = soup.select_one("#chapter-content, .chapter-c")
+    if content is None:
+        raise ValueError(f"Chapter text was not found at {chapter['url']}. The page may be blocked or its layout changed.")
+    for tag in content.select("script, style, iframe, ins, button, nav, .ads, .adsbygoogle, [hidden], h1, h2, h3"):
+        tag.decompose()
+    for tag in content.find_all(["p", "div", "br", "li"]):
+        tag.append(soup.new_string("\n\n"))
+    raw = content.get_text(separator="")
+    raw = _WATERMARK.sub("", raw).replace("\u200b", "")
+    paragraphs = []
+    leading = True
+    title_number = _CHAPTER_HEADING.match(chapter["title"])
+    for block in re.split(r"\n\s*\n", raw):
+        text = " ".join(block.split())
+        if not text:
+            continue
+        heading = _CHAPTER_HEADING.match(text)
+        # Some chapters repeat the title with a different subtitle. Match the
+        # chapter number so the existing workflow can speak the title just once.
+        same_heading = (heading and title_number and heading[1] == title_number[1])
+        if leading and (text.casefold() == chapter["title"].casefold()
+                        or same_heading or _CREDITS.match(text)):
+            continue
+        leading = False
+        paragraphs.append(text)
+    text = "\n\n".join(paragraphs)
+    if not text:
+        raise ValueError(f"Chapter {chapter['title']} has no readable body text.")
+    return {"title": chapter["title"], "text": text}
+
+
+def _load_chapter(chapter, cache_dir, refresh):
+    digest = hashlib.sha256(chapter["url"].encode()).hexdigest()
+    path = cache_dir / "chapters" / f"{digest}.json"
+    cached = _read_cache(path) if not refresh else None
+    if isinstance(cached, dict) and cached.get("text") and cached.get("title"):
+        return cached
+    result = _parse_chapter(_fetch_html(chapter["url"]), chapter)
+    _write_cache(path, result)
+    return result
+
+
+class OmniVoiceWebNovelLoader:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {"required": {
+            "novel_url": ("STRING", {
+                "default": "https://novelfull.net/release-that-witch.html",
+                "tooltip": "NovelFull's novel URL. Downloads the chapter index once and caches selected chapter text locally.",
+            }),
+            "chapter_start": ("INT", {
+                "default": 1, "min": 1, "max": 99999, "step": 1,
+                "tooltip": "First chapter in reading order (1-indexed). Connect your incrementing chapter counter here.",
+            }),
+            "chapter_end": ("INT", {
+                "default": 1, "min": 1, "max": 99999, "step": 1,
+                "tooltip": "Last chapter, inclusive. Connect the same counter as chapter_start for one audio file per chapter.",
+            }),
+            "refresh_cache": ("BOOLEAN", {
+                "default": False,
+                "tooltip": "Re-download the chapter index and selected chapters. Leave off during batch generation.",
+            }),
+        }}
+
+    RETURN_TYPES = ("STRING", "STRING", "STRING")
+    RETURN_NAMES = ("text", "chapter_title", "chapter_list")
+    FUNCTION = "load_novel"
+    CATEGORY = "OmniVoice"
+
+    @classmethod
+    def IS_CHANGED(cls, novel_url, chapter_start, chapter_end, refresh_cache=False):
+        return float("nan") if refresh_cache else False
+
+    def load_novel(self, novel_url, chapter_start, chapter_end, refresh_cache=False):
+        url = _novel_url(novel_url)
+        cache_dir = _CACHE_DIR / hashlib.sha256(url.encode()).hexdigest()[:16]
+        chapters = _load_index(url, cache_dir, refresh_cache)
+        if not 1 <= chapter_start <= len(chapters):
+            raise ValueError(f"Chapter {chapter_start} is outside this novel's range (1–{len(chapters)}). Stop the queue or change the counter.")
+        end = max(chapter_start, min(chapter_end, len(chapters)))
+        selected = []
+        for chapter in chapters[chapter_start - 1:end]:
+            selected.append(_load_chapter(chapter, cache_dir, refresh_cache))
+        text = "\n\n---\n\n".join(ch["text"] for ch in selected)
+        title = selected[0]["title"]
+        chapter_list = "\n".join(f"{i}. {ch['title']}" for i, ch in enumerate(chapters, 1))
+        return text, title, chapter_list