""" Novel Scraper β€” Gradio UI Universal adaptive web novel downloader. """ from __future__ import annotations import asyncio import io import json import os import shutil import threading import time import zipfile from pathlib import Path from typing import Optional import gradio as gr from scraper.engine import ChapterStatus, FetchMode, NovelJob from scraper.manager import DownloadConfig, DownloadManager, load_state from utils.logging_setup import AppLogger from utils.range_parser import parse_chapter_range, format_range_hint # --------------------------------------------------------------------------- # Global state # --------------------------------------------------------------------------- OUTPUT_DIR = Path("output") OUTPUT_DIR.mkdir(parents=True, exist_ok=True) app_logger = AppLogger.get(OUTPUT_DIR) # Active download managers keyed by job URL _managers: dict[str, DownloadManager] = {} _jobs: dict[str, NovelJob] = {} _threads: dict[str, threading.Thread] = {} _results: dict[str, Optional[Path]] = {} _loop: Optional[asyncio.AbstractEventLoop] = None def _get_loop() -> asyncio.AbstractEventLoop: global _loop if _loop is None or _loop.is_closed(): _loop = asyncio.new_event_loop() threading.Thread(target=_loop.run_forever, daemon=True).start() return _loop def _run_async(coro): loop = _get_loop() future = asyncio.run_coroutine_threadsafe(coro, loop) return future # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _parse_cookies(cookie_str: str) -> dict: """Parse 'key=value; key2=value2' or JSON cookie string.""" cookie_str = cookie_str.strip() if not cookie_str: return {} if cookie_str.startswith("{") or cookie_str.startswith("["): try: data = json.loads(cookie_str) if isinstance(data, list): return {item["name"]: item["value"] for item in data if "name" in item} return data except Exception: return {} cookies = {} for part in cookie_str.split(";"): kv = part.strip().split("=", 1) if len(kv) == 2: cookies[kv[0].strip()] = kv[1].strip() return cookies def _build_progress_table(job: NovelJob) -> str: if not job.chapters: return "No chapters discovered yet." done = job.done_count failed = job.failed_count locked = sum(1 for c in job.chapters if c.status == ChapterStatus.LOCKED) skipped = sum(1 for c in job.chapters if c.status == ChapterStatus.SKIPPED) pct = f"{job.progress*100:.1f}" elapsed = "" if job.started_at: e = time.time() - job.started_at elapsed = f"{int(e//60)}m {int(e%60)}s" eta = "" if job.started_at and done > 0 and job.progress < 1: rate = done / max(time.time() - job.started_at, 1) remaining = job.total - done eta_secs = remaining / rate eta = f"{int(eta_secs//60)}m {int(eta_secs%60)}s" return ( f"**{job.title or 'Novel'}**\n\n" f"| Stat | Value |\n|------|-------|\n" f"| Status | {job.status.upper()} |\n" f"| Progress | {done}/{job.total} ({pct}%) |\n" f"| Done | βœ… {done} |\n" f"| Failed | ❌ {failed} |\n" f"| Locked | πŸ’° {locked} |\n" f"| Skipped | ⏭️ {skipped} |\n" f"| Elapsed | {elapsed or 'β€”'} |\n" f"| ETA | {eta or 'β€”'} |\n" ) def _build_failed_list(job: NovelJob) -> str: failed = [c for c in job.chapters if c.status == ChapterStatus.FAILED] if not failed: return "No failed chapters." lines = [f"**{len(failed)} failed chapters:**\n"] for c in failed[:50]: lines.append(f"- [{c.title}]({c.url}) β€” {c.error}") if len(failed) > 50: lines.append(f"... and {len(failed)-50} more (see failed_chapters.txt)") return "\n".join(lines) def _zip_outputs(job: NovelJob) -> Optional[str]: """Create a ZIP of all output files for a job.""" safe_title = "".join(c if c.isalnum() or c in " -_" else "_" for c in (job.title or "novel"))[:60] zip_path = OUTPUT_DIR / f"{safe_title}_package.zip" with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: for f in OUTPUT_DIR.iterdir(): if f.is_file() and not f.name.startswith("."): zf.write(f, f.name) return str(zip_path) # --------------------------------------------------------------------------- # Core download entrypoint (runs in background thread) # --------------------------------------------------------------------------- def _start_download_thread( url: str, config: DownloadConfig, resume: bool = False, ) -> None: """Launch download in background thread.""" # Try to restore state job = None if resume: job = load_state(url, OUTPUT_DIR) if job: app_logger.ui_log(f"πŸ“‚ Resumed from saved state: {job.title}") if job is None: job = NovelJob(url=url, output_dir=OUTPUT_DIR) job.start_chapter = config.__dict__.get("start_chapter", 1) job.end_chapter = config.__dict__.get("end_chapter", 0) job.specific_chapters = config.__dict__.get("specific_chapters", []) _jobs[url] = job def log_cb(msg: str) -> None: app_logger.ui_log(msg) def progress_cb(j: NovelJob) -> None: pass # polling-based UI manager = DownloadManager(job, config, log_cb=log_cb, progress_cb=progress_cb) _managers[url] = manager async def _run(): try: result = await manager.run() _results[url] = result except Exception as exc: app_logger.ui_log(f"❌ Download error: {exc}") _results[url] = None job.status = "failed" future = _run_async(_run()) _results[url] = None # --------------------------------------------------------------------------- # Gradio event handlers # --------------------------------------------------------------------------- def add_novels( urls_text: str, mode_str: str, rate: float, concurrency: int, retries: int, timeout: int, cookie_str: str, chapter_range: str, resume_prev: bool, ) -> tuple[str, str, str]: """Parse inputs and launch downloads.""" urls = [u.strip() for u in urls_text.strip().splitlines() if u.strip()] if not urls: return "⚠️ Please enter at least one URL.", "", "" mode_map = {"Fast": FetchMode.FAST, "Balanced": FetchMode.BALANCED, "Compatibility": FetchMode.COMPAT} mode = mode_map.get(mode_str, FetchMode.BALANCED) cookies = _parse_cookies(cookie_str) # Parse range specific = parse_chapter_range(chapter_range) start_ch = 1 end_ch = 0 if specific: start_ch = min(specific) end_ch = max(specific) config = DownloadConfig( mode=mode, max_concurrent=max(1, min(concurrency, 10)), rate_limit=max(0.1, 1.0 / max(rate, 0.1)), max_retries=max(1, retries), timeout=max(5, timeout), cookies=cookies, output_dir=OUTPUT_DIR, ) config.__dict__["start_chapter"] = start_ch config.__dict__["end_chapter"] = end_ch config.__dict__["specific_chapters"] = specific messages = [] for url in urls: if url in _jobs and _jobs[url].status == "running": messages.append(f"⚠️ Already running: {url}") continue app_logger.ui_log(f"πŸš€ Queuing: {url}") t = threading.Thread( target=_start_download_thread, args=(url, config, resume_prev), daemon=True, ) t.start() _threads[url] = t messages.append(f"βœ… Started: {url}") return "\n".join(messages), "", app_logger.get_ui_text() def pause_job(url: str) -> str: url = url.strip() if url in _managers: _managers[url].pause() return f"⏸️ Paused: {url}" return "⚠️ No active download for that URL." def resume_job(url: str) -> str: url = url.strip() if url in _managers: _managers[url].resume() return f"▢️ Resumed: {url}" return "⚠️ No active download for that URL." def cancel_job(url: str) -> str: url = url.strip() if url in _managers: _managers[url].cancel() return f"🚫 Cancelled: {url}" return "⚠️ No active download for that URL." def retry_failed(url: str) -> str: url = url.strip() if url not in _jobs: return "⚠️ No job found for that URL." job = _jobs[url] for c in job.chapters: if c.status == ChapterStatus.FAILED: c.status = ChapterStatus.PENDING c.retries = 0 c.error = "" app_logger.ui_log(f"πŸ”„ Reset failed chapters for retry: {url}") return f"βœ… Failed chapters reset for retry." def get_status() -> tuple[str, str, str, gr.File]: """Poll status β€” called by timer.""" if not _jobs: return "No downloads started.", "", app_logger.get_ui_text(), None progress_parts = [] failed_parts = [] for url, job in _jobs.items(): progress_parts.append(_build_progress_table(job)) failed_parts.append(_build_failed_list(job)) # Find completed downloads for file output output_file = None for url, result_path in _results.items(): if result_path and Path(result_path).exists(): output_file = str(result_path) break return ( "\n\n---\n\n".join(progress_parts), "\n\n---\n\n".join(failed_parts), app_logger.get_ui_text(200), output_file, ) def get_download_files() -> list[str]: """Return list of available output files.""" files = [] for f in OUTPUT_DIR.iterdir(): if f.is_file() and not f.name.startswith(".") and f.suffix in (".txt", ".zip"): files.append(str(f)) return files def create_zip() -> Optional[str]: if not _jobs: return None # Use the most recent job job = list(_jobs.values())[-1] return _zip_outputs(job) def clear_logs() -> str: app_logger.ring.clear() return "" def preview_extraction(url: str, mode_str: str, cookie_str: str) -> str: """Preview extracted content from a URL without saving.""" url = url.strip() if not url: return "⚠️ Enter a URL to preview." mode_map = {"Fast": FetchMode.FAST, "Balanced": FetchMode.BALANCED, "Compatibility": FetchMode.COMPAT} mode = mode_map.get(mode_str, FetchMode.BALANCED) cookies = _parse_cookies(cookie_str) from scraper.engine import create_session, smart_fetch from scraper.extractor import extract_chapter async def _preview(): session = await create_session(cookies) try: html, status = await smart_fetch(session, url, mode=mode, cookies=cookies) title, content, paras = extract_chapter(html) preview = f"**Title:** {title}\n\n**Preview (first 500 chars):**\n\n{content[:500]}..." return preview finally: await session.close() try: future = _run_async(_preview()) result = future.result(timeout=30) return result except Exception as exc: return f"❌ Preview failed: {exc}" # --------------------------------------------------------------------------- # CSS theme # --------------------------------------------------------------------------- CUSTOM_CSS = """ @import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@400;600&family=Sora:wght@300;400;600;700&display=swap'); :root { --bg-primary: #0d0f14; --bg-secondary: #141720; --bg-card: #1a1e2e; --bg-input: #1e2335; --accent: #7c6af7; --accent-bright: #9d8fff; --accent-dim: #4a3fa0; --green: #22d68a; --red: #ff4d6d; --yellow: #ffc947; --text-primary: #e8ecf8; --text-secondary: #8892b0; --text-muted: #4a5270; --border: #252a3d; --border-bright: #353d5a; --radius: 12px; --radius-sm: 8px; } * { box-sizing: border-box; } body, .gradio-container { background: var(--bg-primary) !important; font-family: 'Sora', sans-serif !important; color: var(--text-primary) !important; } .gradio-container { max-width: 1400px !important; margin: 0 auto !important; padding: 20px !important; } /* Header */ .app-header { text-align: center; padding: 40px 20px 30px; border-bottom: 1px solid var(--border); margin-bottom: 30px; } .app-title { font-size: 2.4rem; font-weight: 700; letter-spacing: -0.5px; background: linear-gradient(135deg, var(--accent-bright), var(--green)); -webkit-background-clip: text; -webkit-text-fill-color: transparent; background-clip: text; margin: 0 0 8px 0; } .app-subtitle { color: var(--text-secondary); font-size: 0.95rem; font-weight: 300; letter-spacing: 0.5px; } /* Tabs */ .tab-nav button { background: var(--bg-card) !important; border: 1px solid var(--border) !important; color: var(--text-secondary) !important; border-radius: var(--radius-sm) !important; font-family: 'Sora', sans-serif !important; font-weight: 500 !important; transition: all 0.2s !important; } .tab-nav button.selected, .tab-nav button:hover { background: var(--accent-dim) !important; color: var(--accent-bright) !important; border-color: var(--accent) !important; } /* Panels / blocks */ .panel-card { background: var(--bg-card); border: 1px solid var(--border); border-radius: var(--radius); padding: 20px; margin-bottom: 16px; } /* Labels */ label, .label-wrap span { color: var(--text-secondary) !important; font-family: 'Sora', sans-serif !important; font-size: 0.82rem !important; font-weight: 600 !important; letter-spacing: 0.4px !important; text-transform: uppercase !important; } /* Inputs */ textarea, input[type="text"], input[type="number"], .gr-input, .gr-textarea { background: var(--bg-input) !important; border: 1px solid var(--border-bright) !important; border-radius: var(--radius-sm) !important; color: var(--text-primary) !important; font-family: 'Sora', sans-serif !important; font-size: 0.9rem !important; } textarea:focus, input:focus { border-color: var(--accent) !important; box-shadow: 0 0 0 2px rgba(124, 106, 247, 0.15) !important; outline: none !important; } /* Buttons */ button.gr-button { font-family: 'Sora', sans-serif !important; font-weight: 600 !important; border-radius: var(--radius-sm) !important; transition: all 0.2s !important; letter-spacing: 0.3px !important; } button.gr-button-primary { background: linear-gradient(135deg, var(--accent), var(--accent-dim)) !important; border: none !important; color: white !important; } button.gr-button-primary:hover { background: linear-gradient(135deg, var(--accent-bright), var(--accent)) !important; transform: translateY(-1px) !important; box-shadow: 0 4px 20px rgba(124, 106, 247, 0.3) !important; } button.gr-button-secondary { background: var(--bg-input) !important; border: 1px solid var(--border-bright) !important; color: var(--text-primary) !important; } button.gr-button-secondary:hover { border-color: var(--accent) !important; color: var(--accent-bright) !important; } /* Sliders */ input[type="range"] { accent-color: var(--accent) !important; } /* Dropdowns */ .gr-dropdown { background: var(--bg-input) !important; border: 1px solid var(--border-bright) !important; border-radius: var(--radius-sm) !important; color: var(--text-primary) !important; } /* Markdown output */ .gr-markdown, .output-markdown { color: var(--text-primary) !important; font-family: 'Sora', sans-serif !important; font-size: 0.9rem !important; line-height: 1.7 !important; } .gr-markdown h1, .gr-markdown h2 { color: var(--accent-bright) !important; } .gr-markdown table { width: 100%; border-collapse: collapse; } .gr-markdown th { background: var(--bg-input); color: var(--text-secondary); padding: 8px 12px; font-size: 0.78rem; letter-spacing: 0.5px; text-transform: uppercase; } .gr-markdown td { padding: 8px 12px; border-bottom: 1px solid var(--border); color: var(--text-primary); } .gr-markdown code { background: var(--bg-input); color: var(--accent-bright); padding: 2px 6px; border-radius: 4px; font-family: 'JetBrains Mono', monospace; font-size: 0.82rem; } /* Logs textarea */ .logs-area textarea { font-family: 'JetBrains Mono', monospace !important; font-size: 0.78rem !important; line-height: 1.6 !important; color: var(--green) !important; background: #080a10 !important; border: 1px solid #1a2040 !important; } /* Progress bar inside gr components */ .progress-bar { background: var(--accent) !important; } /* Status indicators */ .status-badge { display: inline-block; padding: 2px 10px; border-radius: 20px; font-size: 0.72rem; font-weight: 700; text-transform: uppercase; letter-spacing: 0.5px; } /* File download button */ .file-download button { background: var(--green) !important; color: #0d1a0f !important; font-weight: 700 !important; } /* Accordion */ .gr-accordion { background: var(--bg-card) !important; border: 1px solid var(--border) !important; border-radius: var(--radius) !important; } /* Checkboxes */ input[type="checkbox"] { accent-color: var(--accent) !important; } /* Divider */ .gr-separator { border-color: var(--border) !important; } /* Scrollbar */ ::-webkit-scrollbar { width: 6px; height: 6px; } ::-webkit-scrollbar-track { background: var(--bg-primary); } ::-webkit-scrollbar-thumb { background: var(--border-bright); border-radius: 3px; } ::-webkit-scrollbar-thumb:hover { background: var(--accent-dim); } """ # --------------------------------------------------------------------------- # Build Gradio app # --------------------------------------------------------------------------- def build_app() -> gr.Blocks: with gr.Blocks( title="Novel Scraper β€” Universal Downloader", css=CUSTOM_CSS, theme=gr.themes.Base( primary_hue=gr.themes.colors.purple, neutral_hue=gr.themes.colors.slate, font=[gr.themes.GoogleFont("Sora"), "ui-sans-serif"], ), ) as demo: # Header gr.HTML("""
πŸ“– Novel Scraper
Universal adaptive web novel downloader Β· Works on any site
""") with gr.Tabs(): # ---------------------------------------------------------------- # TAB 1: Download # ---------------------------------------------------------------- with gr.Tab("⬇️ Download"): with gr.Row(): # Left: Input panel with gr.Column(scale=2): gr.Markdown("### Novel URLs") urls_input = gr.Textbox( label="Novel URLs (one per line)", placeholder="https://example.com/novel-name\nhttps://another-site.com/story\n...", lines=5, elem_classes=["urls-input"], ) with gr.Row(): chapter_range = gr.Textbox( label="Chapter Range (leave blank for all)", placeholder="e.g. all, 1-100, 1-50,70-90, 5,10,15", lines=1, ) resume_check = gr.Checkbox( label="Resume previous download", value=True, ) with gr.Accordion("βš™οΈ Download Settings", open=False): mode_select = gr.Dropdown( label="Fetch Mode", choices=["Fast", "Balanced", "Compatibility"], value="Balanced", info="Fast=aiohttp only Β· Balanced=+cloudscraper Β· Compatibility=+Playwright", ) with gr.Row(): rate_slider = gr.Slider( label="Chapters per second", minimum=0.1, maximum=10.0, step=0.1, value=1.0, info="Higher = faster (may trigger bans)", ) concurrency = gr.Slider( label="Concurrent downloads", minimum=1, maximum=10, step=1, value=3, ) with gr.Row(): retries = gr.Slider( label="Max retries per chapter", minimum=1, maximum=30, step=1, value=10, ) timeout = gr.Slider( label="Request timeout (seconds)", minimum=5, maximum=120, step=5, value=30, ) with gr.Accordion("πŸͺ Cookies / Authentication", open=False): cookie_input = gr.Textbox( label="Browser Cookies", placeholder='key=value; key2=value2\n\nor paste JSON: [{"name":"...","value":"..."}]', lines=4, info="Paste cookies from browser DevTools for age-gated or login-required sites.", ) with gr.Row(): start_btn = gr.Button("πŸš€ Start Download", variant="primary", scale=2) preview_btn = gr.Button("πŸ‘οΈ Preview", variant="secondary", scale=1) status_msg = gr.Textbox( label="Status", lines=3, interactive=False, show_label=True, ) # Right: Controls panel with gr.Column(scale=1): gr.Markdown("### Queue Controls") control_url = gr.Textbox( label="URL to control", placeholder="Paste the novel URL here", lines=1, ) with gr.Row(): pause_btn = gr.Button("⏸️ Pause", variant="secondary") resume_btn = gr.Button("▢️ Resume", variant="secondary") with gr.Row(): cancel_btn = gr.Button("🚫 Cancel", variant="secondary") retry_btn = gr.Button("πŸ”„ Retry Failed", variant="secondary") control_msg = gr.Textbox( label="Control Output", lines=2, interactive=False, ) gr.Markdown("### Preview") preview_out = gr.Markdown( value="*Click Preview to extract content from a URL.*", label="", ) # ---------------------------------------------------------------- # TAB 2: Progress # ---------------------------------------------------------------- with gr.Tab("πŸ“Š Progress"): refresh_btn = gr.Button("πŸ”„ Refresh", variant="secondary") progress_display = gr.Markdown( value="*No downloads started yet. Go to the Download tab.*", label="", ) gr.Markdown("### Failed Chapters") failed_display = gr.Markdown(value="*None*") # ---------------------------------------------------------------- # TAB 3: Logs # ---------------------------------------------------------------- with gr.Tab("πŸ“‹ Logs"): with gr.Row(): refresh_logs_btn = gr.Button("πŸ”„ Refresh Logs", variant="secondary") clear_logs_btn = gr.Button("πŸ—‘οΈ Clear", variant="secondary") logs_out = gr.Textbox( label="Live Logs", lines=25, interactive=False, max_lines=25, elem_classes=["logs-area"], value="", ) # ---------------------------------------------------------------- # TAB 4: Downloads # ---------------------------------------------------------------- with gr.Tab("πŸ’Ύ Download Files"): gr.Markdown("### Available Output Files") list_files_btn = gr.Button("πŸ” List Files", variant="secondary") zip_btn = gr.Button("πŸ“¦ Create ZIP Package", variant="secondary") files_display = gr.Markdown(value="*Click 'List Files' to see outputs.*") file_download = gr.File( label="Download File", interactive=False, elem_classes=["file-download"], ) # ---------------------------------------------------------------- # TAB 5: Help # ---------------------------------------------------------------- with gr.Tab("❓ Help"): gr.Markdown(""" ## How to Use ### 1. Basic Usage 1. Paste one or more novel URLs (one per line) in the **Novel URLs** box 2. Optionally set a chapter range (e.g. `1-100` or `all`) 3. Click **Start Download** 4. Monitor progress in the **Progress** tab 5. Download your `.txt` file from the **Download Files** tab ### 2. Chapter Ranges | Expression | Meaning | |-----------|---------| | *(blank)* or `all` | All chapters | | `1-100` | Chapters 1 through 100 | | `50-200` | Chapters 50 through 200 | | `1,5,9,20` | Specific chapters only | | `1-50,70-90` | Multiple ranges | ### 3. Fetch Modes | Mode | Description | |------|------------| | **Fast** | aiohttp only β€” fastest, may fail on protected sites | | **Balanced** | aiohttp + cloudscraper fallback β€” recommended | | **Compatibility** | + Playwright browser β€” works on most sites, slower | ### 4. Cookies For sites requiring login or age verification: 1. Open the site in your browser and log in 2. Open DevTools β†’ Application β†’ Cookies 3. Copy cookie pairs and paste in the Cookies box 4. Or export as JSON from a cookie manager extension ### 5. Output Format The merged TXT file uses simple clean formatting: ``` Chapter Title Chapter content paragraphs... Next Chapter Title Next content... ``` ### 6. Failed Chapters If chapters fail after all retries, they appear in `failed_chapters.txt`. Use the **Retry Failed** button after a pause to attempt redownload. ### 7. Resume Downloads save state automatically. Check **Resume previous download** to continue an interrupted session. --- ### Supported Structures - Direct chapter lists on the main page - Paginated chapter lists (`?page=N`, `/page/N`) - Chapter links in subpages (`/chapters`, `/toc`, `/episodes`) - JSON embedded in `