""" 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("""