LouisMylle/eboek.info-scraper
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 3 Wiki Activity
Files
9f548309f82f7775d924c0de2a1a4e6f251b6a84
eboek.info-scraper/core/bookys_scraper.py
Louis Mylle c6a69c6ada feat: Add Bookys scraper alongside EBoek with startup mode chooser 
Adds a second scraper for bookys-ebooks.com that harvests direct 1fichier
download URLs into a .txt file. The existing EBoek scraper core is unchanged;
only the GUI gained a Mode menu entry to switch between the two.

Bookys specifics:
- Visible (never headless) browser, since the site is behind Cloudflare.
  The challenge is auto-detected: the scraper polls for real listing items
  and continues on its own once it clears, no button needed.
- Persistent Chrome profile in ~/.eboek_scraper/bookys_chrome_profile so the
  Cloudflare clearance cookie survives between runs. Removed the hardcoded
  user-agent override, which claimed Chrome/120 against a real Chrome/150 and
  invalidated that cookie on every launch.
- Pop-up ads are avoided by never clicking host links: hrefs are read and
  navigated to directly, with window.open neutered via CDP as a backstop.
- Host links are read with textContent rather than Selenium's .text, which
  returns empty for these elements because they sit in a collapsed container.
- Only 1fichier hosts are followed; other hosts on a page are ignored.

Build configs list the new modules as hidden imports in all four places.
gui_main imports the Bookys window lazily, so PyInstaller's static analysis
would otherwise miss it and the .exe would crash on switching modes.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NVJc2XuTvqBHSuqotetseS
2026-07-19 20:05:49 +02:00

452 lines
18 KiB
Python
Raw Blame History

"""
Bookys-ebooks.com scraper.
Unlike the EBoek scraper (which downloads files), this scraper harvests the
direct 1fichier.com download URLs for each item in a category and writes them,
one per line, to a plain text file.
Key differences from the EBoek scraper:
* Never runs headless - the site is behind Cloudflare and the user must pass
the "I am not a robot" challenge manually. The scraper auto-detects when the
challenge clears and continues on its own.
* Never clicks host/download links. The site spawns pop-up ads on click, so we
read the href and navigate to it directly (driver.get). Pop-ups are also
neutralised by overriding window.open on every document.
* Only follows 1fichier hosts (per requirements), ignoring other file hosts.
"""
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
import time
import random
from pathlib import Path
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Category slug -> listing path on the site. Only BD is enabled for now.
CATEGORIES = {
"bd": {
"name": "Bandes dessinées (BD)",
"path": "bandes-dessinees/bd",
},
}
BASE_DOMAIN = "https://www7.bookys-ebooks.com"
# Persistent Chrome profile for the Bookys scraper. Keeping this between runs is
# what stops Cloudflare from re-challenging on every launch.
DEFAULT_PROFILE_DIR = Path.home() / ".eboek_scraper" / "bookys_chrome_profile"
class BookysScraper:
"""Harvests 1fichier download URLs from bookys-ebooks.com category listings."""
# Selectors, kept together so a future site change is easy to patch.
ITEM_SELECTOR = ".bys-items-container a.bys-item"
HOST_LINK_SELECTOR = "a.bys-link.bys-host"
def __init__(self, progress_callback=None, category="bd", output_file=None,
timing_config=None, profile_dir=None):
"""
Args:
progress_callback (callable): callback(event_type: str, data: dict)
category (str): category slug (see CATEGORIES)
output_file (str|Path): path to the .txt file to write URLs into
timing_config (dict): optional timing overrides
profile_dir (str|Path): persistent Chrome profile directory. Keeping a
profile between runs preserves the Cloudflare clearance cookie, so
the challenge doesn't reappear on every launch.
"""
self.progress_callback = progress_callback
self._stop_requested = False
self.category = category if category in CATEGORIES else "bd"
self.output_file = Path(output_file) if output_file else (
Path.home() / "Downloads" / "bookys_1fichier_links.txt"
)
# Dedicated profile - deliberately NOT the user's everyday Chrome profile,
# which Chrome refuses to open while a normal Chrome window is running.
self.profile_dir = Path(profile_dir) if profile_dir else DEFAULT_PROFILE_DIR
self.timing = timing_config or {}
self._setup_timing_defaults()
# Collected URLs (also written to disk as we go) and a dedupe set.
self.collected_urls = []
self._seen_urls = set()
# Popup-neutralising script injected into every new document.
self._popup_kill_js = "window.open = function(){ return null; };"
chrome_options = Options()
# NOTE: never headless - Cloudflare needs a real, visible browser.
# Persistent profile: this is what keeps the Cloudflare clearance cookie
# alive between runs, so the challenge isn't shown on every single launch.
try:
self.profile_dir.mkdir(parents=True, exist_ok=True)
except Exception:
pass
chrome_options.add_argument(f'--user-data-dir={self.profile_dir}')
chrome_options.add_argument('--profile-directory=Default')
chrome_options.add_argument('--ignore-ssl-errors')
chrome_options.add_argument('--ignore-certificate-errors')
chrome_options.add_argument('--disable-blink-features=AutomationControlled')
chrome_options.add_experimental_option("excludeSwitches",
["enable-automation", "enable-logging"])
chrome_options.add_experimental_option('useAutomationExtension', False)
chrome_options.add_argument('--log-level=3')
# No user-agent override on purpose: a spoofed UA that disagrees with the
# real Chrome build invalidates the clearance cookie and re-triggers the
# challenge. Letting Chrome send its genuine UA is both safer and stealthier.
try:
self.driver = webdriver.Chrome(options=chrome_options)
except Exception as e:
# Most common cause: the profile is already locked by another run.
self._emit("scraper_init_failed", {
"error": str(e),
"profile_dir": str(self.profile_dir),
"hint": "If a previous scraper Chrome window is still open, close it "
"and try again (the profile can only be used by one at a time).",
})
raise
self.driver.execute_script(
"Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
)
# Kill window.open before any page script runs, on every navigation.
try:
self.driver.execute_cdp_cmd(
'Page.addScriptToEvaluateOnNewDocument',
{'source': self._popup_kill_js}
)
except Exception:
# Non-Chromium drivers won't support CDP; per-page fallback still runs.
pass
self._emit("scraper_initialized", {"category": self.category,
"output_file": str(self.output_file)})
# ------------------------------------------------------------------ helpers
def _emit(self, event_type, data):
if self.progress_callback:
try:
self.progress_callback(event_type, data)
except Exception:
pass
def request_stop(self):
"""Ask the scraper to stop gracefully at the next checkpoint."""
self._stop_requested = True
self._emit("stop_requested", {})
def _setup_timing_defaults(self):
defaults = {
'action_delay_min': 0.8,
'action_delay_max': 2.5,
'page_break_min': 3,
'page_break_max': 8,
}
for key, value in defaults.items():
self.timing.setdefault(key, value)
def _delay(self, min_sec=None, max_sec=None):
if self._stop_requested:
return
if min_sec is None:
min_sec = self.timing['action_delay_min']
if max_sec is None:
max_sec = self.timing['action_delay_max']
time.sleep(random.uniform(min_sec, max_sec))
@staticmethod
def _link_text(element):
"""
Read an element's text via textContent rather than Selenium's .text.
The host links on Bookys live in a collapsed container, so they report as
not-displayed and .text returns an empty string for all of them.
textContent reads the DOM directly and works regardless of visibility.
"""
try:
return (element.get_attribute("textContent") or "").strip()
except Exception:
return ""
def _kill_popups(self):
"""Override window.open on the current document and close any stray tabs."""
try:
self.driver.execute_script(self._popup_kill_js)
except Exception:
pass
self._close_extra_tabs()
def _close_extra_tabs(self):
"""Close any tab that isn't the main one (defensive - ads can slip a tab in)."""
try:
handles = self.driver.window_handles
if len(handles) <= 1:
return
main = handles[0]
for handle in handles[1:]:
try:
self.driver.switch_to.window(handle)
self.driver.close()
except Exception:
pass
self.driver.switch_to.window(main)
except Exception:
pass
def _navigate(self, url):
"""Navigate directly to a URL and neutralise pop-ups afterwards."""
if self._stop_requested:
return False
self.driver.get(url)
self._kill_popups()
return True
def _listing_url(self, page_num):
path = CATEGORIES[self.category]["path"]
base = f"{BASE_DOMAIN}/{path}"
if page_num <= 1:
return base
return f"{base}?page={page_num}"
# --------------------------------------------------------------- cloudflare
def wait_for_content(self, timeout=600, poll_interval=2):
"""
Wait until the category listing renders real items, giving the user time
to clear the Cloudflare challenge. Auto-detects success - no button.
Returns True once items are found, False on timeout/stop.
"""
self._emit("cloudflare_check", {"message": "Waiting for Cloudflare / page to load..."})
deadline = time.time() + timeout
announced = False
while time.time() < deadline:
if self._stop_requested:
return False
try:
items = self.driver.find_elements(By.CSS_SELECTOR, self.ITEM_SELECTOR)
if items:
self._emit("cloudflare_passed", {"item_count": len(items)})
return True
except Exception:
pass
if not announced:
# Only nudge the user once, when content isn't immediately present.
self._emit("cloudflare_waiting", {
"message": "If a Cloudflare check is shown, please tick the "
"checkbox in the browser. Scraping continues automatically."
})
announced = True
time.sleep(poll_interval)
self._emit("cloudflare_timeout", {"timeout": timeout})
return False
# -------------------------------------------------------------------- write
def _record_url(self, url, context):
if not url or "1fichier" not in url:
return False
if url in self._seen_urls:
self._emit("link_duplicate", {"url": url, **context})
return False
self._seen_urls.add(url)
self.collected_urls.append(url)
try:
self.output_file.parent.mkdir(parents=True, exist_ok=True)
with open(self.output_file, "a", encoding="utf-8") as f:
f.write(url + "\n")
except Exception as e:
self._emit("write_error", {"error": str(e), "url": url})
self._emit("link_found", {"url": url, "total": len(self.collected_urls), **context})
return True
# ------------------------------------------------------------------- detail
def _process_book(self, book_url, page_num, book_index, total_books):
"""Open a book detail page and harvest its 1fichier link(s)."""
self._emit("book_started", {
"url": book_url, "page_number": page_num,
"book_index": book_index, "total_books": total_books,
})
if not self._navigate(book_url):
return 0
self._delay()
title = ""
try:
title = self.driver.find_element(By.CSS_SELECTOR, "h1").text.strip()
except Exception:
pass
# Collect intermediate /dl/<id> hrefs for 1fichier hosts only.
# A page typically lists several hosts (1fichier, DailyUploads, Filefox...);
# everything except 1fichier is deliberately ignored.
dl_hrefs = []
hosts_seen = []
try:
host_links = self.driver.find_elements(By.CSS_SELECTOR, self.HOST_LINK_SELECTOR)
if not host_links:
# Fallback if the site changes its classes: any /dl/ link.
host_links = self.driver.find_elements(By.CSS_SELECTOR, "a[href*='/dl/']")
for link in host_links:
try:
text = self._link_text(link)
if text:
hosts_seen.append(text)
if "1fichier" in text.lower():
href = link.get_attribute("href")
if href:
dl_hrefs.append(href)
except Exception:
continue
except Exception as e:
self._emit("book_error", {"url": book_url, "error": str(e)})
return 0
self._emit("hosts_listed", {"url": book_url, "hosts": hosts_seen,
"fichier_count": len(dl_hrefs)})
found = 0
for dl_href in dl_hrefs:
if self._stop_requested:
break
# The /dl/ page carries the direct 1fichier URL as an anchor href.
if not self._navigate(dl_href):
continue
self._delay()
final_url = None
try:
# Primary: any anchor already pointing at 1fichier. This is the
# most direct signal and doesn't depend on the button's wording.
for a in self.driver.find_elements(By.CSS_SELECTOR, "a[href*='1fichier']"):
href = a.get_attribute("href")
if href:
final_url = href
break
# Fallback: locate the call-to-action by its text (textContent,
# not .text - the element may report as not displayed).
if not final_url:
for a in self.driver.find_elements(By.TAG_NAME, "a"):
try:
text = self._link_text(a).lower()
if "cliquez ici" in text or "chargement" in text:
href = a.get_attribute("href")
if href and "1fichier" in href:
final_url = href
break
except Exception:
continue
except Exception as e:
self._emit("book_error", {"url": dl_href, "error": str(e)})
if final_url and self._record_url(final_url, {"title": title, "book_url": book_url}):
found += 1
self._emit("book_completed", {
"url": book_url, "title": title, "links_found": found,
"page_number": page_num, "book_index": book_index,
})
return found
# -------------------------------------------------------------------- scrape
def scrape(self, start_page=1, end_page=1):
"""Walk pages start_page..end_page and harvest 1fichier URLs."""
if self._stop_requested:
return {"success": False, "reason": "Cancelled before starting"}
total_pages = end_page - start_page + 1
self._emit("scraping_started", {
"start_page": start_page, "end_page": end_page,
"total_pages": total_pages,
"category": CATEGORIES[self.category]["name"],
"output_file": str(self.output_file),
})
errors = []
pages_done = 0
for page_num in range(start_page, end_page + 1):
if self._stop_requested:
break
page_url = self._listing_url(page_num)
self._emit("page_started", {
"page_number": page_num,
"page_index": page_num - start_page + 1,
"total_pages": total_pages,
"url": page_url,
})
if not self._navigate(page_url):
continue
# First page load may show Cloudflare; wait it out (auto-detect).
if not self.wait_for_content():
errors.append(f"Timed out waiting for content on page {page_num}")
self._emit("page_error", {"page_number": page_num,
"error": "content did not load"})
break
try:
book_urls = [a.get_attribute("href") for a
in self.driver.find_elements(By.CSS_SELECTOR, self.ITEM_SELECTOR)]
book_urls = [u for u in book_urls if u]
except Exception as e:
errors.append(f"Page {page_num}: {e}")
self._emit("page_error", {"page_number": page_num, "error": str(e)})
continue
self._emit("page_items_found", {"page_number": page_num,
"item_count": len(book_urls)})
for i, book_url in enumerate(book_urls, 1):
if self._stop_requested:
break
try:
self._process_book(book_url, page_num, i, len(book_urls))
except Exception as e:
errors.append(f"Book {book_url}: {e}")
self._emit("book_error", {"url": book_url, "error": str(e)})
self._delay()
pages_done += 1
self._emit("page_completed", {"page_number": page_num,
"items_processed": len(book_urls)})
# Short break between pages.
if page_num < end_page and not self._stop_requested:
time.sleep(random.uniform(self.timing['page_break_min'],
self.timing['page_break_max']))
summary = {
"success": not self._stop_requested and not errors,
"cancelled": self._stop_requested,
"total_pages_processed": pages_done,
"total_links_found": len(self.collected_urls),
"output_file": str(self.output_file),
"errors": errors,
}
self._emit("scraping_completed", summary)
return summary
def close(self):
try:
self.driver.quit()
self._emit("scraper_closed", {})
except Exception as e:
self._emit("scraper_close_error", {"error": str(e)})
Reference in New Issue View Git Blame Copy Permalink