diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..c91f494 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,23 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + - name: Install dependencies + run: pip install -r requirements-dev.txt + - name: Lint + run: ruff check . && ruff format --check . + - name: Run tests + run: pytest -v diff --git a/README.md b/README.md index ba8d28e..9625788 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,7 @@ **A complete web scraping kit with BeautifulSoup, Selenium, and Playwright** +[![CI](https://github.com/LacerdaTraderCode/web-scraper-toolkit/actions/workflows/ci.yml/badge.svg)](https://github.com/LacerdaTraderCode/web-scraper-toolkit/actions/workflows/ci.yml) [![Python](https://img.shields.io/badge/Python-3.11%2B-blue?logo=python&logoColor=white)](https://www.python.org/) [![Selenium](https://img.shields.io/badge/Selenium-43B02A?logo=selenium&logoColor=white)](https://www.selenium.dev/) [![Playwright](https://img.shields.io/badge/Playwright-2EAD33?logo=playwright&logoColor=white)](https://playwright.dev/) @@ -57,7 +58,10 @@ web-scraper-toolkit/ │ ├── quotes_bs4.py # Scraping quotes.toscrape.com │ ├── books_selenium.py # Scraping books.toscrape.com │ └── news_playwright.py # News scraping (async) +├── tests/ +├── .github/workflows/ci.yml ├── requirements.txt +├── requirements-dev.txt └── README.md ``` @@ -85,19 +89,19 @@ playwright install chromium ### BeautifulSoup — static HTML ```bash -python examples/quotes_bs4.py +python -m examples.quotes_bs4 ``` Extracts quotes from `quotes.toscrape.com`, saves to CSV and Parquet. ### Selenium — JavaScript-heavy sites ```bash -python examples/books_selenium.py +python -m examples.books_selenium ``` Navigates the `books.toscrape.com` catalog simulating a real user. ### Playwright — asynchronous scraping ```bash -python examples/news_playwright.py +python -m examples.news_playwright ``` Collects headlines asynchronously (faster than Selenium). @@ -113,6 +117,18 @@ Collects headlines asynchronously (faster than Selenium). --- +## 🧪 Development + +```bash +pip install -r requirements-dev.txt +ruff check . && ruff format --check . +pytest -v +``` + +Browser drivers and HTTP sessions are mocked, so the test suite needs no browser or network access. The same checks run on every push and pull request via GitHub Actions. + +--- + ## ⚖️ Ethical Use This toolkit is for educational purposes. When scraping: diff --git a/examples/books_selenium.py b/examples/books_selenium.py index a774a2a..13c2272 100644 --- a/examples/books_selenium.py +++ b/examples/books_selenium.py @@ -1,18 +1,11 @@ -""" -Exemplo: scraping de livros usando Selenium. -Executar: python examples/books_selenium.py -""" -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent)) +import logging from scrapers.selenium_scraper import SeleniumScraper from utils.exporters import export_all def main(): - print("🚀 Iniciando scraping com Selenium...\n") + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") scraper = SeleniumScraper(headless=True) try: @@ -21,8 +14,6 @@ def main(): finally: scraper.close() - print("\n✅ Concluído!") - if __name__ == "__main__": main() diff --git a/examples/news_playwright.py b/examples/news_playwright.py index e046bb4..dddacdc 100644 --- a/examples/news_playwright.py +++ b/examples/news_playwright.py @@ -1,26 +1,17 @@ -""" -Exemplo: scraping assíncrono com Playwright. -Executar: python examples/news_playwright.py -""" import asyncio -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).parent.parent)) +import logging from scrapers.playwright_scraper import PlaywrightScraper from utils.exporters import export_all async def main(): - print("🚀 Iniciando scraping assíncrono com Playwright...\n") + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") scraper = PlaywrightScraper() quotes = await scraper.scrape_quotes_async(max_pages=5) export_all(quotes, "output/quotes_playwright") - print("\n✅ Concluído!") - if __name__ == "__main__": asyncio.run(main()) diff --git a/examples/quotes_bs4.py b/examples/quotes_bs4.py index 5e244eb..9d96a26 100644 --- a/examples/quotes_bs4.py +++ b/examples/quotes_bs4.py @@ -1,19 +1,11 @@ -""" -Exemplo: scraping de quotes usando BeautifulSoup. -Executar: python examples/quotes_bs4.py -""" -import sys -from pathlib import Path - -# Adiciona raiz do projeto ao path -sys.path.insert(0, str(Path(__file__).parent.parent)) +import logging from scrapers.bs4_scraper import BeautifulSoupScraper from utils.exporters import export_all def main(): - print("🚀 Iniciando scraping com BeautifulSoup...\n") + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") scraper = BeautifulSoupScraper(min_interval=1.0) try: @@ -22,8 +14,6 @@ def main(): finally: scraper.close() - print("\n✅ Concluído!") - if __name__ == "__main__": main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e3e67d8 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,6 @@ +[tool.ruff] +line-length = 100 +exclude = [".venv"] + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "RUF"] diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..636340f --- /dev/null +++ b/pytest.ini @@ -0,0 +1,4 @@ +[pytest] +python_files = test_*.py +testpaths = tests +asyncio_mode = auto diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..61a1097 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,4 @@ +-r requirements.txt +pytest==9.1.1 +pytest-asyncio==1.4.0 +ruff==0.16.9 diff --git a/scrapers/bs4_scraper.py b/scrapers/bs4_scraper.py index 539d3c2..bb10566 100644 --- a/scrapers/bs4_scraper.py +++ b/scrapers/bs4_scraper.py @@ -1,17 +1,15 @@ -""" -Scraper usando BeautifulSoup + Requests para HTML estático. -""" +import logging + import requests from bs4 import BeautifulSoup from tenacity import retry, stop_after_attempt, wait_exponential -from typing import List, Dict from utils.rate_limiter import RateLimiter, get_random_user_agent +logger = logging.getLogger(__name__) -class BeautifulSoupScraper: - """Scraper simples e rápido para páginas estáticas.""" +class BeautifulSoupScraper: def __init__(self, min_interval: float = 1.0): self.rate_limiter = RateLimiter(min_interval) self.session = requests.Session() @@ -19,9 +17,9 @@ def __init__(self, min_interval: float = 1.0): @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), + reraise=True, ) def fetch(self, url: str) -> BeautifulSoup: - """Busca uma URL e retorna objeto BeautifulSoup.""" self.rate_limiter.wait() headers = {"User-Agent": get_random_user_agent()} @@ -30,41 +28,35 @@ def fetch(self, url: str) -> BeautifulSoup: return BeautifulSoup(response.text, "lxml") - def scrape_quotes(self, base_url: str = "https://quotes.toscrape.com") -> List[Dict]: - """ - Exemplo: extrai quotes do site quotes.toscrape.com (site oficial pra aprender). - """ + @staticmethod + def _parse_quote(div) -> dict: + tags = [tag.get_text(strip=True) for tag in div.find_all("a", class_="tag")] + return { + "text": div.find("span", class_="text").get_text(strip=True), + "author": div.find("small", class_="author").get_text(strip=True), + "tags": ", ".join(tags), + } + + def scrape_quotes(self, base_url: str = "https://quotes.toscrape.com") -> list[dict]: quotes = [] page = 1 while True: - url = f"{base_url}/page/{page}/" - print(f"📄 Scraping página {page}...") - - soup = self.fetch(url) + logger.info("Scraping page %d", page) + soup = self.fetch(f"{base_url}/page/{page}/") quote_divs = soup.find_all("div", class_="quote") if not quote_divs: break - for div in quote_divs: - quotes.append({ - "text": div.find("span", class_="text").get_text(strip=True), - "author": div.find("small", class_="author").get_text(strip=True), - "tags": ", ".join( - tag.get_text(strip=True) - for tag in div.find_all("a", class_="tag") - ), - }) - - # Verifica se tem próxima página - next_btn = soup.find("li", class_="next") - if not next_btn: + quotes.extend(self._parse_quote(div) for div in quote_divs) + + if not soup.find("li", class_="next"): break page += 1 - print(f"✅ Total coletado: {len(quotes)} quotes") + logger.info("Collected %d quotes", len(quotes)) return quotes - def close(self): + def close(self) -> None: self.session.close() diff --git a/scrapers/playwright_scraper.py b/scrapers/playwright_scraper.py index a0fae78..4e46c99 100644 --- a/scrapers/playwright_scraper.py +++ b/scrapers/playwright_scraper.py @@ -1,65 +1,54 @@ -""" -Scraper assíncrono usando Playwright - moderno e rápido. -""" import asyncio -from typing import List, Dict -from playwright.async_api import async_playwright +import logging +from playwright.async_api import Error, async_playwright -class PlaywrightScraper: - """Scraper moderno, assíncrono e multi-browser.""" +logger = logging.getLogger(__name__) +EXTRACT_QUOTES_SCRIPT = """ +() => { + const items = document.querySelectorAll('.quote'); + return Array.from(items).map(q => ({ + text: q.querySelector('.text')?.innerText || '', + author: q.querySelector('.author')?.innerText || '', + tags: Array.from(q.querySelectorAll('.tag')).map(t => t.innerText).join(', ') + })); +} +""" + + +class PlaywrightScraper: async def scrape_quotes_async( self, base_url: str = "https://quotes.toscrape.com", max_pages: int = 5 - ) -> List[Dict]: - """ - Scraping assíncrono - várias páginas em paralelo. - Muito mais rápido que Selenium. - """ + ) -> list[dict]: quotes = [] - async with async_playwright() as p: - browser = await p.chromium.launch(headless=True) - context = await browser.new_context() - - # Criar tasks em paralelo - tasks = [] - for page_num in range(1, max_pages + 1): - tasks.append( - self._scrape_page(context, f"{base_url}/page/{page_num}/") - ) - - results = await asyncio.gather(*tasks, return_exceptions=True) - - for result in results: - if isinstance(result, list): - quotes.extend(result) - - await browser.close() - - print(f"✅ Total coletado: {len(quotes)} quotes (async)") + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(headless=True) + try: + context = await browser.new_context() + tasks = [ + self._scrape_page(context, f"{base_url}/page/{page_number}/") + for page_number in range(1, max_pages + 1) + ] + results = await asyncio.gather(*tasks, return_exceptions=True) + finally: + await browser.close() + + for result in results: + if isinstance(result, list): + quotes.extend(result) + + logger.info("Collected %d quotes (async)", len(quotes)) return quotes - async def _scrape_page(self, context, url: str) -> List[Dict]: - """Extrai quotes de uma única página.""" + async def _scrape_page(self, context, url: str) -> list[dict]: page = await context.new_page() try: await page.goto(url, wait_until="domcontentloaded", timeout=15000) - - quotes = await page.evaluate(""" - () => { - const items = document.querySelectorAll('.quote'); - return Array.from(items).map(q => ({ - text: q.querySelector('.text')?.innerText || '', - author: q.querySelector('.author')?.innerText || '', - tags: Array.from(q.querySelectorAll('.tag')) - .map(t => t.innerText).join(', ') - })); - } - """) - return quotes - except Exception as e: - print(f"⚠️ Erro em {url}: {e}") + return await page.evaluate(EXTRACT_QUOTES_SCRIPT) + except Error as exc: + logger.warning("Failed to scrape %s: %s", url, exc) return [] finally: await page.close() diff --git a/scrapers/selenium_scraper.py b/scrapers/selenium_scraper.py index ef617af..e77c210 100644 --- a/scrapers/selenium_scraper.py +++ b/scrapers/selenium_scraper.py @@ -1,82 +1,77 @@ -""" -Scraper usando Selenium para sites com JavaScript legado. -""" -from typing import List, Dict +import logging + from selenium import webdriver +from selenium.common.exceptions import NoSuchElementException, TimeoutException from selenium.webdriver.chrome.options import Options +from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By +from selenium.webdriver.support import expected_conditions as ec from selenium.webdriver.support.ui import WebDriverWait -from selenium.webdriver.support import expected_conditions as EC from webdriver_manager.chrome import ChromeDriverManager -from selenium.webdriver.chrome.service import Service from utils.rate_limiter import get_random_user_agent +logger = logging.getLogger(__name__) + +PRODUCT_LOCATOR = (By.CLASS_NAME, "product_pod") + + +def create_driver(headless: bool = True) -> webdriver.Chrome: + options = Options() + if headless: + options.add_argument("--headless=new") + options.add_argument("--no-sandbox") + options.add_argument("--disable-dev-shm-usage") + options.add_argument(f"user-agent={get_random_user_agent()}") + + service = Service(ChromeDriverManager().install()) + return webdriver.Chrome(service=service, options=options) + class SeleniumScraper: - """Scraper para sites que dependem de JavaScript.""" - - def __init__(self, headless: bool = True): - options = Options() - if headless: - options.add_argument("--headless=new") - options.add_argument("--no-sandbox") - options.add_argument("--disable-dev-shm-usage") - options.add_argument(f"user-agent={get_random_user_agent()}") - - service = Service(ChromeDriverManager().install()) - self.driver = webdriver.Chrome(service=service, options=options) + def __init__(self, headless: bool = True, driver=None): + self.driver = driver or create_driver(headless) self.wait = WebDriverWait(self.driver, 10) - def scrape_books(self, base_url: str = "https://books.toscrape.com") -> List[Dict]: - """ - Exemplo: extrai livros do catálogo do site books.toscrape.com. - """ + @staticmethod + def _parse_book(article) -> dict: + rating_class = article.find_element(By.CLASS_NAME, "star-rating").get_attribute("class") + return { + "title": article.find_element(By.CSS_SELECTOR, "h3 a").get_attribute("title"), + "price": article.find_element(By.CLASS_NAME, "price_color").text, + "stock": article.find_element(By.CLASS_NAME, "availability").text.strip(), + "rating": rating_class.replace("star-rating ", ""), + } + + def _go_to_next_page(self) -> bool: + try: + self.driver.find_element(By.CSS_SELECTOR, "li.next a").click() + self.wait.until(ec.presence_of_all_elements_located(PRODUCT_LOCATOR)) + except (NoSuchElementException, TimeoutException): + return False + return True + + def scrape_books( + self, base_url: str = "https://books.toscrape.com", max_pages: int = 3 + ) -> list[dict]: books = [] self.driver.get(base_url) + self.wait.until(ec.presence_of_all_elements_located(PRODUCT_LOCATOR)) - # Aguarda carregamento - self.wait.until( - EC.presence_of_all_elements_located((By.CLASS_NAME, "product_pod")) - ) - - pages_scraped = 0 - while pages_scraped < 3: # Limite de páginas para demo - print(f"📄 Processando página {pages_scraped + 1}...") + for page_number in range(1, max_pages + 1): + logger.info("Processing page %d", page_number) - articles = self.driver.find_elements(By.CLASS_NAME, "product_pod") - for article in articles: + for article in self.driver.find_elements(*PRODUCT_LOCATOR): try: - books.append({ - "title": article.find_element( - By.CSS_SELECTOR, "h3 a" - ).get_attribute("title"), - "price": article.find_element( - By.CLASS_NAME, "price_color" - ).text, - "stock": article.find_element( - By.CLASS_NAME, "availability" - ).text.strip(), - "rating": article.find_element( - By.CLASS_NAME, "star-rating" - ).get_attribute("class").replace("star-rating ", ""), - }) - except Exception as e: - print(f"⚠️ Erro extraindo livro: {e}") - - # Próxima página - try: - next_btn = self.driver.find_element(By.CSS_SELECTOR, "li.next a") - next_btn.click() - self.wait.until( - EC.presence_of_all_elements_located((By.CLASS_NAME, "product_pod")) - ) - pages_scraped += 1 - except Exception: + books.append(self._parse_book(article)) + except NoSuchElementException as exc: + logger.warning("Failed to extract book: %s", exc) + + if page_number == max_pages or not self._go_to_next_page(): break - print(f"✅ Total coletado: {len(books)} livros") + logger.info("Collected %d books", len(books)) return books - def close(self): + def close(self) -> None: self.driver.quit() diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_bs4_scraper.py b/tests/test_bs4_scraper.py new file mode 100644 index 0000000..a223e83 --- /dev/null +++ b/tests/test_bs4_scraper.py @@ -0,0 +1,87 @@ +from unittest.mock import MagicMock + +import pytest +import requests +from bs4 import BeautifulSoup +from tenacity import wait_none + +from scrapers import bs4_scraper +from scrapers.bs4_scraper import BeautifulSoupScraper + +PAGE_ONE = """ +
+ First quote + Author A +
lifelove
+
+ +""" + +PAGE_TWO = """ +
+ Second quote + Author B +
+""" + + +def soup(html): + return BeautifulSoup(html, "lxml") + + +@pytest.fixture +def scraper(monkeypatch): + monkeypatch.setattr(bs4_scraper, "get_random_user_agent", lambda: "Test-Agent") + instance = BeautifulSoupScraper() + instance.rate_limiter = MagicMock() + instance.session = MagicMock() + return instance + + +@pytest.fixture(autouse=True) +def no_retry_delay(monkeypatch): + monkeypatch.setattr(BeautifulSoupScraper.fetch.retry, "wait", wait_none()) + + +def test_fetch_returns_parsed_html(scraper): + scraper.session.get.return_value.text = PAGE_TWO + + result = scraper.fetch("https://example.com") + + assert result.find("span", class_="text").get_text() == "Second quote" + scraper.rate_limiter.wait.assert_called_once_with() + _, kwargs = scraper.session.get.call_args + assert kwargs["headers"] == {"User-Agent": "Test-Agent"} + + +def test_fetch_retries_then_raises_http_error(scraper): + scraper.session.get.return_value.raise_for_status.side_effect = requests.HTTPError("500") + + with pytest.raises(requests.HTTPError): + scraper.fetch("https://example.com") + + assert scraper.session.get.call_count == 3 + + +def test_scrape_quotes_follows_pagination(scraper, monkeypatch): + pages = {"/page/1/": PAGE_ONE, "/page/2/": PAGE_TWO} + monkeypatch.setattr(scraper, "fetch", lambda url: soup(pages[url.removeprefix("http://t")])) + + quotes = scraper.scrape_quotes(base_url="http://t") + + assert quotes == [ + {"text": "First quote", "author": "Author A", "tags": "life, love"}, + {"text": "Second quote", "author": "Author B", "tags": ""}, + ] + + +def test_scrape_quotes_stops_on_empty_page(scraper, monkeypatch): + monkeypatch.setattr(scraper, "fetch", lambda url: soup("")) + + assert scraper.scrape_quotes(base_url="http://t") == [] + + +def test_close_closes_session(scraper): + scraper.close() + + scraper.session.close.assert_called_once_with() diff --git a/tests/test_exporters.py b/tests/test_exporters.py new file mode 100644 index 0000000..e2c350d --- /dev/null +++ b/tests/test_exporters.py @@ -0,0 +1,49 @@ +import polars as pl + +from utils.exporters import export_all, export_to_csv, export_to_json, export_to_parquet + +DATA = [ + {"text": "First quote", "author": "Author A", "tags": "life, love"}, + {"text": "Second quote", "author": "Author B", "tags": "books"}, +] + + +def test_export_to_csv_roundtrip(tmp_path): + path = tmp_path / "quotes.csv" + + export_to_csv(DATA, str(path)) + + assert pl.read_csv(path).to_dicts() == DATA + + +def test_export_to_parquet_roundtrip(tmp_path): + path = tmp_path / "quotes.parquet" + + export_to_parquet(DATA, str(path)) + + assert pl.read_parquet(path).to_dicts() == DATA + + +def test_export_to_json_writes_data(tmp_path): + path = tmp_path / "quotes.json" + + export_to_json(DATA, str(path)) + + content = path.read_text(encoding="utf-8") + assert "First quote" in content + assert "Author B" in content + + +def test_export_creates_missing_directories(tmp_path): + path = tmp_path / "nested" / "output" / "quotes.csv" + + export_to_csv(DATA, str(path)) + + assert path.exists() + + +def test_export_all_writes_every_format(tmp_path): + export_all(DATA, str(tmp_path / "out" / "quotes")) + + produced = sorted(file.name for file in (tmp_path / "out").iterdir()) + assert produced == ["quotes.csv", "quotes.json", "quotes.parquet"] diff --git a/tests/test_playwright_scraper.py b/tests/test_playwright_scraper.py new file mode 100644 index 0000000..6883086 --- /dev/null +++ b/tests/test_playwright_scraper.py @@ -0,0 +1,93 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest +from playwright.async_api import Error + +from scrapers import playwright_scraper +from scrapers.playwright_scraper import PlaywrightScraper + + +def make_page(quotes=None, error=None): + page = MagicMock() + page.goto = AsyncMock(side_effect=error) + page.evaluate = AsyncMock(return_value=quotes if quotes is not None else []) + page.close = AsyncMock() + return page + + +class FakePlaywrightManager: + def __init__(self, browser): + self.playwright = MagicMock() + self.playwright.chromium.launch = AsyncMock(return_value=browser) + + async def __aenter__(self): + return self.playwright + + async def __aexit__(self, *exc_info): + return False + + +@pytest.fixture +def browser(): + instance = MagicMock() + instance.new_context = AsyncMock(return_value=MagicMock()) + instance.close = AsyncMock() + return instance + + +@pytest.fixture +def install_browser(monkeypatch, browser): + monkeypatch.setattr( + playwright_scraper, + "async_playwright", + lambda: FakePlaywrightManager(browser), + ) + return browser + + +async def test_scrape_page_returns_extracted_quotes(): + quote = {"text": "Q", "author": "A", "tags": "t"} + page = make_page(quotes=[quote]) + context = MagicMock() + context.new_page = AsyncMock(return_value=page) + + result = await PlaywrightScraper()._scrape_page(context, "http://t/page/1/") + + assert result == [quote] + page.goto.assert_awaited_once() + page.close.assert_awaited_once() + + +async def test_scrape_page_returns_empty_list_on_playwright_error(): + page = make_page(error=Error("navigation failed")) + context = MagicMock() + context.new_page = AsyncMock(return_value=page) + + result = await PlaywrightScraper()._scrape_page(context, "http://t/page/1/") + + assert result == [] + page.close.assert_awaited_once() + + +async def test_scrape_quotes_async_combines_pages_in_order(install_browser): + first = {"text": "One", "author": "A", "tags": ""} + second = {"text": "Two", "author": "B", "tags": ""} + context = await install_browser.new_context() + context.new_page = AsyncMock(side_effect=[make_page([first]), make_page([second])]) + + quotes = await PlaywrightScraper().scrape_quotes_async(base_url="http://t", max_pages=2) + + assert quotes == [first, second] + install_browser.close.assert_awaited_once() + + +async def test_scrape_quotes_async_ignores_failed_pages(install_browser): + quote = {"text": "Only", "author": "A", "tags": ""} + context = await install_browser.new_context() + context.new_page = AsyncMock( + side_effect=[make_page(error=Error("timeout")), make_page([quote])], + ) + + quotes = await PlaywrightScraper().scrape_quotes_async(base_url="http://t", max_pages=2) + + assert quotes == [quote] diff --git a/tests/test_rate_limiter.py b/tests/test_rate_limiter.py new file mode 100644 index 0000000..f19980b --- /dev/null +++ b/tests/test_rate_limiter.py @@ -0,0 +1,70 @@ +import pytest + +from utils import rate_limiter +from utils.rate_limiter import FALLBACK_USER_AGENT, RateLimiter, get_random_user_agent + + +@pytest.fixture +def clock(monkeypatch): + state = {"now": 100.0, "slept": []} + + def fake_sleep(seconds): + state["slept"].append(seconds) + state["now"] += seconds + + monkeypatch.setattr(rate_limiter.time, "monotonic", lambda: state["now"]) + monkeypatch.setattr(rate_limiter.time, "sleep", fake_sleep) + return state + + +def test_first_request_does_not_wait(clock): + RateLimiter(min_interval=1.0).wait() + + assert clock["slept"] == [] + + +def test_consecutive_requests_wait_for_full_interval(clock): + limiter = RateLimiter(min_interval=1.0) + + limiter.wait() + limiter.wait() + + assert clock["slept"] == [pytest.approx(1.0)] + + +def test_wait_only_covers_remaining_interval(clock): + limiter = RateLimiter(min_interval=1.0) + limiter.wait() + clock["now"] += 0.4 + + limiter.wait() + + assert clock["slept"] == [pytest.approx(0.6)] + + +def test_no_wait_after_interval_elapsed(clock): + limiter = RateLimiter(min_interval=1.0) + limiter.wait() + clock["now"] += 5 + + limiter.wait() + + assert clock["slept"] == [] + + +def test_random_user_agent_comes_from_library(monkeypatch): + class FakeUserAgent: + random = "Test-Agent/1.0" + + monkeypatch.setattr(rate_limiter, "UserAgent", FakeUserAgent) + + assert get_random_user_agent() == "Test-Agent/1.0" + + +def test_random_user_agent_falls_back_on_failure(monkeypatch): + def broken(): + raise RuntimeError("no network") + + monkeypatch.setattr(rate_limiter, "UserAgent", broken) + + assert get_random_user_agent() == FALLBACK_USER_AGENT diff --git a/tests/test_selenium_scraper.py b/tests/test_selenium_scraper.py new file mode 100644 index 0000000..191230d --- /dev/null +++ b/tests/test_selenium_scraper.py @@ -0,0 +1,120 @@ +from unittest.mock import MagicMock + +import pytest +from selenium.common.exceptions import NoSuchElementException, TimeoutException + +from scrapers import selenium_scraper +from scrapers.selenium_scraper import SeleniumScraper + + +def make_article(title="Book", price="£10.00", stock=" In stock ", rating="star-rating Three"): + elements = { + "star-rating": MagicMock(**{"get_attribute.return_value": rating}), + "h3 a": MagicMock(**{"get_attribute.return_value": title}), + "price_color": MagicMock(text=price), + "availability": MagicMock(text=stock), + } + article = MagicMock() + article.find_element.side_effect = lambda by, value: elements[value] + return article + + +@pytest.fixture +def driver(): + return MagicMock() + + +@pytest.fixture +def scraper(driver): + instance = SeleniumScraper(driver=driver) + instance.wait = MagicMock() + return instance + + +def test_parse_book_extracts_fields(): + article = make_article("Dune", "£9.99", " In stock ", "star-rating Five") + + book = SeleniumScraper._parse_book(article) + + assert book == {"title": "Dune", "price": "£9.99", "stock": "In stock", "rating": "Five"} + + +def test_scrape_books_single_page(scraper, driver): + driver.find_elements.return_value = [make_article("A"), make_article("B")] + + books = scraper.scrape_books(base_url="http://t", max_pages=1) + + assert [book["title"] for book in books] == ["A", "B"] + driver.get.assert_called_once_with("http://t") + driver.find_element.assert_not_called() + + +def test_scrape_books_follows_next_page(scraper, driver): + driver.find_elements.side_effect = [[make_article("A")], [make_article("B")]] + + books = scraper.scrape_books(max_pages=2) + + assert [book["title"] for book in books] == ["A", "B"] + driver.find_element.return_value.click.assert_called_once_with() + + +def test_scrape_books_stops_without_next_link(scraper, driver): + driver.find_elements.return_value = [make_article("A")] + driver.find_element.side_effect = NoSuchElementException("no next link") + + books = scraper.scrape_books(max_pages=3) + + assert len(books) == 1 + + +def test_scrape_books_stops_when_next_page_times_out(scraper, driver): + driver.find_elements.return_value = [make_article("A")] + scraper.wait.until.side_effect = [None, TimeoutException()] + + books = scraper.scrape_books(max_pages=3) + + assert len(books) == 1 + + +def test_scrape_books_skips_unparseable_items(scraper, driver): + broken = MagicMock() + broken.find_element.side_effect = NoSuchElementException("missing") + driver.find_elements.return_value = [broken, make_article("Valid")] + + books = scraper.scrape_books(max_pages=1) + + assert [book["title"] for book in books] == ["Valid"] + + +def test_close_quits_driver(scraper, driver): + scraper.close() + + driver.quit.assert_called_once_with() + + +def test_constructor_creates_driver_when_not_injected(monkeypatch): + factory = MagicMock() + monkeypatch.setattr(selenium_scraper, "create_driver", factory) + + scraper = SeleniumScraper(headless=False) + + assert scraper.driver is factory.return_value + factory.assert_called_once_with(False) + + +@pytest.mark.parametrize("headless", [True, False]) +def test_create_driver_configures_chrome_options(monkeypatch, headless): + chrome = MagicMock() + manager = MagicMock() + manager.return_value.install.return_value = "/driver" + monkeypatch.setattr(selenium_scraper.webdriver, "Chrome", chrome) + monkeypatch.setattr(selenium_scraper, "ChromeDriverManager", manager) + monkeypatch.setattr(selenium_scraper, "Service", MagicMock()) + monkeypatch.setattr(selenium_scraper, "get_random_user_agent", lambda: "Test-Agent") + + driver = selenium_scraper.create_driver(headless=headless) + + assert driver is chrome.return_value + arguments = chrome.call_args.kwargs["options"].arguments + assert ("--headless=new" in arguments) is headless + assert "user-agent=Test-Agent" in arguments diff --git a/utils/exporters.py b/utils/exporters.py index bdff4d1..62c51c4 100644 --- a/utils/exporters.py +++ b/utils/exporters.py @@ -1,40 +1,35 @@ -""" -Exportadores de dados para múltiplos formatos usando Polars. -""" +import logging from pathlib import Path -from typing import List, Dict + import polars as pl +logger = logging.getLogger(__name__) + -def export_to_csv(data: List[Dict], filepath: str) -> None: - """Exporta lista de dicts para CSV usando Polars (mais rápido que Pandas).""" - df = pl.DataFrame(data) +def _to_frame(data: list[dict], filepath: str) -> pl.DataFrame: Path(filepath).parent.mkdir(parents=True, exist_ok=True) - df.write_csv(filepath) - print(f"✅ CSV salvo: {filepath} ({len(df)} linhas)") + return pl.DataFrame(data) -def export_to_json(data: List[Dict], filepath: str) -> None: - """Exporta lista de dicts para JSON.""" - df = pl.DataFrame(data) - Path(filepath).parent.mkdir(parents=True, exist_ok=True) - df.write_json(filepath) - print(f"✅ JSON salvo: {filepath} ({len(df)} linhas)") +def export_to_csv(data: list[dict], filepath: str) -> None: + frame = _to_frame(data, filepath) + frame.write_csv(filepath) + logger.info("CSV saved: %s (%d rows)", filepath, len(frame)) -def export_to_parquet(data: List[Dict], filepath: str) -> None: - """ - Exporta para Parquet - formato colunar eficiente. - Ideal para grandes volumes de dados e análises posteriores. - """ - df = pl.DataFrame(data) - Path(filepath).parent.mkdir(parents=True, exist_ok=True) - df.write_parquet(filepath, compression="snappy") - print(f"✅ Parquet salvo: {filepath} ({len(df)} linhas, compressão snappy)") +def export_to_json(data: list[dict], filepath: str) -> None: + frame = _to_frame(data, filepath) + frame.write_json(filepath) + logger.info("JSON saved: %s (%d rows)", filepath, len(frame)) + + +def export_to_parquet(data: list[dict], filepath: str) -> None: + frame = _to_frame(data, filepath) + frame.write_parquet(filepath, compression="snappy") + logger.info("Parquet saved: %s (%d rows)", filepath, len(frame)) -def export_all(data: List[Dict], base_path: str) -> None: - """Exporta nos 3 formatos de uma vez.""" +def export_all(data: list[dict], base_path: str) -> None: export_to_csv(data, f"{base_path}.csv") export_to_json(data, f"{base_path}.json") export_to_parquet(data, f"{base_path}.parquet") diff --git a/utils/rate_limiter.py b/utils/rate_limiter.py index 4dea3aa..c17ba4e 100644 --- a/utils/rate_limiter.py +++ b/utils/rate_limiter.py @@ -1,33 +1,29 @@ -""" -Rate limiter simples para evitar sobrecarga de servidores. -""" import time + from fake_useragent import UserAgent +FALLBACK_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" +) -class RateLimiter: - """Controla o intervalo mínimo entre requisições.""" +class RateLimiter: def __init__(self, min_interval: float = 1.0): self.min_interval = min_interval - self._last_request = 0.0 + self._last_request: float | None = None - def wait(self): - """Aguarda se necessário para respeitar o intervalo.""" - elapsed = time.time() - self._last_request - if elapsed < self.min_interval: - time.sleep(self.min_interval - elapsed) - self._last_request = time.time() + def wait(self) -> None: + if self._last_request is not None: + remaining = self.min_interval - (time.monotonic() - self._last_request) + if remaining > 0: + time.sleep(remaining) + self._last_request = time.monotonic() def get_random_user_agent() -> str: - """Retorna um User-Agent aleatório para rotação.""" try: return UserAgent().random except Exception: - # Fallback para UA padrão - return ( - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/120.0.0.0 Safari/537.36" - ) + return FALLBACK_USER_AGENT