Update app.py

app.py

@@ -1,296 +1,120 @@
-"""
-Link Scraper — Hugging Face Space
-----------------------------------
-Upload a .txt file containing one URL per line.
-The app will:
-  1. Visit each URL one by one (or with a few parallel workers)
-  2. Save the raw webpage as an .html file
-  3. Extract the page <title> and meta description
-  4. Produce one combined .txt file with: URL | Title | Description
-  5. Zip all the saved .html files for download
-
-Built for Gradio on Hugging Face Spaces.
-"""
-
-import os
-import re
-import io
+import gradio as gr
+import requests
+from bs4 import BeautifulSoup
 import time
+import os
 import zipfile
-import random
-import concurrent.futures
 from urllib.parse import urlparse
+import re
 
-import requests
-from bs4 import BeautifulSoup
-import gradio as gr
-
-# ---------------------------------------------------------------------------
-# Config
-# ---------------------------------------------------------------------------
-
-USER_AGENT = (
-    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
-    "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
-)
-
-WORK_DIR = "/tmp/link_scraper"
-HTML_DIR = os.path.join(WORK_DIR, "html_pages")
-os.makedirs(HTML_DIR, exist_ok=True)
-
-
-def slugify_for_filename(url: str, index: int) -> str:
-    """Turn a URL into a safe, short filename."""
-    parsed = urlparse(url)
-    path = (parsed.netloc + parsed.path).strip("/")
-    path = re.sub(r"[^a-zA-Z0-9._-]+", "_", path)
-    path = path[:120]  # keep filenames from getting too long
-    if not path:
-        path = "page"
-    return f"{index:05d}_{path}.html"
-
-
-def extract_synopsis(description: str) -> str:
-    """WebNovel-style descriptions read like:
-    'Read X Online for Free, written by ... and the synopsis is: <actual synopsis>'
-    Strip everything before the marker so only the real synopsis remains.
-    If the marker isn't found (non-WebNovel page), just return the description as-is.
-    """
-    marker = "the synopsis is:"
-    idx = description.lower().find(marker)
-    if idx != -1:
-        return description[idx + len(marker):].strip()
-    return description.strip()
-
-
-def extract_title_and_description(html: str) -> tuple[str, str]:
-    """Pull the best available title and synopsis/description from a page's HTML.
-
-    For WebNovel pages the meta description is deliberately truncated with '...'.
-    The full synopsis lives inside a <div class="... j_synopsis ..."> in the body.
-    We prefer that element; the meta tags are only a fallback for other sites.
-    """
-    soup = BeautifulSoup(html, "html.parser")
-
-    # --- Title ---
-    title = ""
-    og_title = soup.find("meta", property="og:title")
-    if og_title and og_title.get("content"):
-        title = og_title["content"].strip()
-    if not title and soup.title and soup.title.string:
-        title = soup.title.string.strip()
-
-    # --- Synopsis: try full body element first (WebNovel and similar sites) ---
-    synopsis = ""
-
-    # 1. WebNovel: <div class="... j_synopsis ...">
-    j_syn = soup.find(class_=lambda c: c and "j_synopsis" in c.split())
-    if j_syn:
-        # Replace <br> tags with newlines before extracting text
-        for br in j_syn.find_all("br"):
-            br.replace_with("\n")
-        # Remove the "read more" toggle span that WebNovel appends
-        for rm in j_syn.find_all("span", class_="_readmore"):
-            rm.decompose()
-        synopsis = j_syn.get_text(separator=" ").strip()
-
-    # 2. Fallback: any element whose class contains common synopsis keywords
-    if not synopsis:
-        for cls_hint in ("book-intro", "book_intro", "bookIntro",
-                         "detail-desc", "book-desc", "bookDesc",
-                         "synopsis", "j_intro"):
-            el = soup.find(class_=lambda c: c and cls_hint in c)
-            if el:
-                for br in el.find_all("br"):
-                    br.replace_with("\n")
-                synopsis = el.get_text(separator=" ").strip()
-                if synopsis:
-                    break
-
-    # 3. Last resort: meta tags (always truncated on WebNovel, but better than nothing)
-    if not synopsis:
-        meta_desc = soup.find("meta", attrs={"name": "description"})
-        if meta_desc and meta_desc.get("content"):
-            synopsis = extract_synopsis(meta_desc["content"].strip())
-    if not synopsis:
-        og_desc = soup.find("meta", property="og:description")
-        if og_desc and og_desc.get("content"):
-            synopsis = extract_synopsis(og_desc["content"].strip())
-
-    # Normalise whitespace: collapse runs of spaces/tabs but keep newlines
-    title = re.sub(r"[ \t]+", " ", title).strip()
-    synopsis = re.sub(r"[ \t]+", " ", synopsis).strip()
-
-    return title, synopsis
-
-
-def fetch_one(index: int, url: str, save_html: bool, timeout: int):
-    """Fetch a single URL. Returns a result dict; never raises."""
-    result = {
-        "index": index,
-        "url": url,
-        "title": "",
-        "description": "",
-        "status": "",
-        "html_filename": "",
-    }
-    try:
-        resp = requests.get(
-            url,
-            headers={"User-Agent": USER_AGENT},
-            timeout=timeout,
-        )
-        result["status"] = f"HTTP {resp.status_code}"
-        resp.raise_for_status()
-        html = resp.text
-
-        if save_html:
-            filename = slugify_for_filename(url, index)
-            filepath = os.path.join(HTML_DIR, filename)
-            with open(filepath, "w", encoding="utf-8") as f:
-                f.write(html)
-            result["html_filename"] = filename
-
-        title, description = extract_title_and_description(html)
-        result["title"] = title
-        result["description"] = description
-
-    except Exception as e:  # noqa: BLE001
-        result["status"] = f"ERROR: {e}"
-
-    return result
-
-
-def zip_html_folder(zip_path: str):
-    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
-        for fname in os.listdir(HTML_DIR):
-            zf.write(os.path.join(HTML_DIR, fname), arcname=fname)
-
-
-def run_scraper(
-    file_obj,
-    save_html: bool,
-    workers: int,
-    delay_seconds: float,
-    timeout_seconds: int,
-    progress=gr.Progress(track_tqdm=False),
-):
-    """Main generator driving the Gradio UI: yields (log_text, combined_file, zip_file)."""
-
-    if file_obj is None:
-        yield "Please upload a .txt file with one URL per line.", None, None
-        return
-
-    # Reset working folders for a clean run
-    if os.path.exists(HTML_DIR):
-        for f in os.listdir(HTML_DIR):
+def scrape_kemono(links_text, delay_sec, max_retries):
+    # Smart URL extraction: find all kemono.cr links even if wrapped in HTML tags
+    raw_urls = re.findall(r'(https://kemono\.cr/[^\s"\'<>]+)', links_text)
+    
+    # Deduplicate while preserving order
+    seen = set()
+    links = [url for url in raw_urls if not (url in seen or seen.add(url))]
+    
+    if not links:
+        return None, "No valid kemono.cr links found in the input."
+    
+    combined_text_path = "combined_extracted_text.txt"
+    html_zip_path = "saved_raw_htmls.zip"
+    
+    extracted_data = []
+    
+    # Create a zip file to store all the individual HTML pages
+    with zipfile.ZipFile(html_zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
+        for i, link in enumerate(links):
             try:
-                os.remove(os.path.join(HTML_DIR, f))
-            except OSError:
-                pass
-    os.makedirs(HTML_DIR, exist_ok=True)
-
-    with open(file_obj, "r", encoding="utf-8", errors="ignore") as f:
-        urls = [line.strip() for line in f if line.strip()]
-
-    total = len(urls)
-    if total == 0:
-        yield "The uploaded file has no URLs in it.", None, None
-        return
-
-    log_lines = [f"Found {total} URLs. Starting...\n"]
-    yield "".join(log_lines), None, None
-
-    results = [None] * total
-    workers = max(1, min(int(workers), 16))
-
-    def worker_task(i, url):
-        if delay_seconds > 0:
-            time.sleep(delay_seconds * random.uniform(0.5, 1.5))
-        return fetch_one(i, url, save_html, int(timeout_seconds))
-
-    completed = 0
-    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
-        futures = {
-            executor.submit(worker_task, i, url): i for i, url in enumerate(urls)
-        }
-        for future in concurrent.futures.as_completed(futures):
-            i = futures[future]
-            res = future.result()
-            results[i] = res
-            completed += 1
-
-            short_url = res["url"][:70]
-            log_lines.append(f"[{completed}/{total}] {res['status']} - {short_url}")
-            progress(completed / total, desc=f"{completed}/{total} pages done")
-
-            # Stream log updates periodically (every 5 items or near the end)
-            if completed % 5 == 0 or completed == total:
-                yield "\n".join(log_lines[-200:]), None, None
-
-    # Build combined output file
-    combined_path = os.path.join(WORK_DIR, "combined_results.txt")
-    with open(combined_path, "w", encoding="utf-8") as f:
-        for res in results:
-            f.write(f"URL: {res['url']}\n")
-            f.write(f"Title: {res['title']}\n")
-            f.write(f"Synopsis: {res['description']}\n")
-            f.write(f"Status: {res['status']}\n")
-            f.write("-" * 60 + "\n")
-
-    zip_path = None
-    if save_html:
-        zip_path = os.path.join(WORK_DIR, "saved_html_pages.zip")
-        zip_html_folder(zip_path)
-
-    ok_count = sum(1 for r in results if r["status"].startswith("HTTP 2"))
-    log_lines.append(
-        f"\nDone. {ok_count}/{total} pages fetched successfully. "
-        f"Combined file and {'zip ' if save_html else ''}ready for download below."
-    )
-
-    yield "\n".join(log_lines[-200:]), combined_path, zip_path
-
-
-with gr.Blocks(title="Link Scraper") as demo:
-    gr.Markdown(
-        "# Link Scraper\n"
-        "Upload a `.txt` file with one URL per line. The app visits each link, "
-        "saves the page as HTML, pulls out the title + description, and gives you "
-        "back one combined `.txt` file (plus a `.zip` of all the saved HTML pages)."
-    )
-
-    with gr.Row():
-        file_input = gr.File(label="Upload .txt file of URLs", file_types=[".txt"])
-
-    with gr.Row():
-        save_html_cb = gr.Checkbox(label="Save raw HTML files (zipped)", value=True)
-        workers_slider = gr.Slider(
-            label="Parallel workers (lower = safer, less likely to get blocked)",
-            minimum=1, maximum=16, step=1, value=4,
-        )
-        delay_slider = gr.Slider(
-            label="Delay between requests per worker (seconds)",
-            minimum=0, maximum=10, step=0.5, value=1,
-        )
-        timeout_slider = gr.Slider(
-            label="Timeout per request (seconds)",
-            minimum=5, maximum=60, step=5, value=20,
-        )
-
-    start_btn = gr.Button("Start Scraping", variant="primary")
-
-    log_box = gr.Textbox(label="Progress / Log", lines=20, max_lines=20)
-
+                # Request Delay Controller
+                if i > 0:
+                    time.sleep(delay_sec)
+                
+                headers = {
+                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+                }
+                
+                # Fetch page with retry mechanism
+                response = None
+                for attempt in range(int(max_retries)):
+                    response = requests.get(link, headers=headers, timeout=15)
+                    if response.status_code == 200:
+                        break
+                    time.sleep(2) # brief pause before retry
+                
+                if response is None or response.status_code != 200:
+                    extracted_data.append(f"--- FAILED TO FETCH: {link} (Status: {response.status_code if response else 'Timeout'}) ---")
+                    continue
+                
+                html_content = response.text
+                
+                # Save Raw HTML to the ZIP file
+                parsed_url = urlparse(link)
+                path_parts = [p for p in parsed_url.path.strip('/').split('/') if p]
+                # Create a safe filename using the last two parts of the URL (e.g., user ID and post ID)
+                safe_name = f"post_{i+1}_" + "_".join(path_parts[-2:]) + ".html"
+                zipf.writestr(safe_name, html_content)
+                
+                # Parse HTML for Text and Title extraction
+                soup = BeautifulSoup(html_content, 'html.parser')
+                
+                # Extract Title (Kemono specifically uses h1 with class 'post__title')
+                title_tag = soup.find('h1', class_='post__title')
+                title = title_tag.get_text(strip=True) if title_tag else f"Untitled_Post_{i+1}"
+                
+                # Extract Content Text (Kemono uses 'post__content' or 'post__body')
+                content_tag = soup.find('div', class_='post__content')
+                if not content_tag:
+                    content_tag = soup.find('div', class_='post__body')
+                
+                text_content = content_tag.get_text(separator='\n', strip=True) if content_tag else "[No text content found in this post.]"
+                
+                # Append formatted data to our combined text string
+                post_text = f"========================================================================\n"
+                post_text += f"TITLE: {title}\n"
+                post_text += f"URL: {link}\n"
+                post_text += f"========================================================================\n\n"
+                post_text += f"{text_content}\n\n\n"
+                
+                extracted_data.append(post_text)
+                
+            except Exception as e:
+                extracted_data.append(f"--- ERROR PROCESSING {link}: {str(e)} ---\n\n")
+
+    # Write the combined text file
+    with open(combined_text_path, 'w', encoding='utf-8') as f:
+        f.write("\n".join(extracted_data))
+        
+    return [combined_text_path, html_zip_path], f"Successfully processed {len(links)} links."
+
+# --- Gradio UI Layout ---
+with gr.Blocks(theme=gr.themes.Soft()) as app:
+    gr.Markdown("# 🦊 Kemono.cr Batch Scraper & Extractor")
+    gr.Markdown("Paste a list of links (or raw HTML tags from your `links.txt` file). The scraper will fetch each link, extract the title and text, save the raw HTMLs in a ZIP, and give you a combined TXT document.")
+    
     with gr.Row():
-        combined_output = gr.File(label="Combined results (.txt)")
-        zip_output = gr.File(label="All saved pages (.zip)")
-
+        with gr.Column(scale=2):
+            links_input = gr.Textbox(
+                lines=12, 
+                label="Paste Links Here", 
+                placeholder="https://kemono.cr/patreon/user/.../post/...\n(You can also paste raw <a href=...> tags)"
+            )
+            
+            with gr.Row():
+                delay_input = gr.Slider(minimum=0, maximum=15, value=2, step=0.5, label="Delay Between Requests (Seconds)")
+                retry_input = gr.Slider(minimum=1, maximum=5, value=3, step=1, label="Max Retries per Link")
+                
+            start_btn = gr.Button("🚀 Start Extraction", variant="primary")
+            
+        with gr.Column(scale=1):
+            status_output = gr.Textbox(label="Status Logging")
+            files_output = gr.File(label="📥 Download Extracted Files")
+            
     start_btn.click(
-        fn=run_scraper,
-        inputs=[file_input, save_html_cb, workers_slider, delay_slider, timeout_slider],
-        outputs=[log_box, combined_output, zip_output],
+        fn=scrape_kemono,
+        inputs=[links_input, delay_input, retry_input],
+        outputs=[files_output, status_output]
     )
 
-if __name__ == "__main__":
-    demo.queue().launch()
+app.launch()