Análisis SEO de la competencia con Python

analisis competencia seo python

Python y SEO forman un dúo ideal: con unas pocas líneas de código es posible reunir en minutos decenas de artículos bien posicionados, extraer su texto y descubrir los temas y palabras clave que están impulsando a la competencia.

1. Esto es lo que vamos a hacer con Python

  1. Obtener los enlaces de las SERP – Podemos hacerlo mediante una API (p. ej. SerpAPI o Google Custom Search) o “scrapeando” Google directamente; la primera opción suele ser más estable pero conlleva coste.
  2. Visitar cada URL – Se emplea requests, una librería HTTP sencilla y robusta que soporta GET, POST y manejo de encoding de forma automática.
  3. Parsear el HTML – BeautifulSoup genera un árbol DOM navegable; es idóneo para extraer títulos, párrafos o listas con selectores CSS.
  4. Persistir y analizar el corpus – Una vez almacenado el texto en bruto podemos aplicar TF-IDF, LLM embeddings o simplemente revisar manualmente los vacíos semánticos.
  5. Escalar y robustecer – Cuando el volumen crece, frameworks como Scrapy gestionan colas de peticiones, respetan robots.txt y aplican auto-throttling.

Tip: Google puede devolver errores 403 si detecta tráfico no humano; encabezados de User-Agent realistas y rotación de IP/proxies minimizan el bloqueo.

2. Prompt 1 – Generar la app con SerAPI

Si te sirve, puedes utilizar este prompt para generar la aplicación:

Rol deseado de la IA  
====================  
Eres un **Ingeniero/a de Datos SEO sénior** que aplica buenas prácticas de desarrollo
(Python ≥ 3.9, PEP 8, PEP 257, tipado estático) y optimización de rendimiento.

Objetivo del script (un solo fichero .py)  
-----------------------------------------
Crear un programa de línea de comandos que:

1. **Recupere** con SerpAPI los *n* primeros resultados orgánicos (por defecto 10) de
   una consulta proporcionada por el usuario.
2. **Descargue** cada URL con `requests` empleando:
   - *timeout*, cabecera *User-Agent* personalizada y reintentos exponenciales ante fallos.
   - Soporte opcional de descarga concurrente (asyncio o `concurrent.futures`).
3. **Extraiga** con `BeautifulSoup` el primer `<h1>`, todos los `<h2>` y los párrafos de
   texto “principales” (heurística: longitud > 40 caracteres y no vacíos).
4. **Guarde** un único fichero JSON UTF-8 con la estructura:
   ```json
   [
     {
       "url":   "...",
       "title": "...",        // <h1>
       "h2":    ["...", "..."],
       "text":  ["párrafo1", "párrafo2", …]
     },
     …
   ]
````

El nombre del archivo y la carpeta de salida deben ser configurables.

## Menú / UI (CLI)

Implementa un menú de opciones con **`argparse`** (o **`click`** si lo prefieres) que permita:

* `--query` (posicional): cadena de búsqueda obligatoria.
* `--results`, `-n`: número de resultados (1-50, por defecto 10).
* `--outfile`, `-o`: ruta del JSON de salida (por defecto `results.json`).
* `--concurrent/--no-concurrent`: activa o desactiva la descarga paralela.
* `--verbose`, `-v`: nivel de *logging* (INFO/DEBUG).
* `--version`: muestra la versión del script.

## Robustez y buenas prácticas

* **Tipado** con *type hints* y comprobación opcional con *mypy*.
* **Docstrings** estilo *Google* o *NumPy*.
* **Logging** en vez de `print`.
* Captura y manejo fino de excepciones (HTTP 4xx/5xx, SSL, SerpAPIError…).
* Separación lógica en funciones y, si procede, clases; evita la programación monolítica.
* Usa `if __name__ == "__main__":` para el *entry-point*.

## Variables de entorno requeridas

* `SERPAPI_KEY`  → API Key de SerpAPI (obligatoria).
* `REQUESTS_UA` → *User-Agent* HTTP (opcional, valor por defecto razonable).
* `HTTP_TIMEOUT` → timeout global en segundos (opcional, defecto 10).

## Instalación y ejecución (añadir al *docstring* principal)

```bash
# 1. Crear entorno virtual
python -m venv .venv && source .venv/bin/activate        # Linux/macOS
#   o bien  .venv\Scripts\activate                       # Windows

# 2. Instalar dependencias mínimas (versión fijada)
pip install -r requirements.txt

# 3. Exportar la API-Key
export SERPAPI_KEY="tu_clave_real_aquí"

# 4. Ejecutar
python seo_scraper.py --query "mejores auriculares inalámbricos" -n 10 -o auriculares.json
```

## Fichero *requirements.txt* sugerido

```
requests>=2.32
beautifulsoup4>=4.12
serpapi>=0.2
tqdm>=4.68          # barra de progreso opcional
```

## Calidad y estilo

* Usa `ruff` o `black` para formateo automático.
* Incorpora comentarios aclaratorios solo cuando aporten contexto extra; evita redundancias.
* Incluye tests unitarios mínimos (doctest o `pytest`) **dentro** del mismo fichero bajo el
  bloque `if __name__ == "__main__":` para no requerir estructura de paquetes.

## Criterio de aceptación

El script generado debe ejecutarse sin errores con la orden de ejemplo anterior y producir
un JSON válido, cumpliendo la estructura y los parámetros especificados.

Este es un posible código para esta aplicación. 100% funcional

#!/usr/bin/env python3
# seo_scraper.py

"""
SEO Scraper CLI

Este script busca en Google usando SerpAPI, descarga las páginas resultantes,
extrae el primer <h1>, todos los <h2> y párrafos de texto principales (>40 caracteres),
y guarda todo en un único JSON UTF-8.

Instalación y ejecución:

    # 1. Crear entorno virtual
    python -m venv .venv && source .venv/bin/activate        # Linux/macOS
    #    o bien  .venv\Scripts\activate                       # Windows

    # 2. Instalar dependencias mínimas
    pip install -r requirements.txt

    # 3. Exportar la API-Key
    export SERPAPI_KEY="tu_clave_real_aquí"

    # Opcional: definir User-Agent y timeout
    export REQUESTS_UA="MiScraper/1.0 (+https://miweb.example)"
    export HTTP_TIMEOUT="15"

    # 4. Ejecutar
    python seo_scraper.py --query "mejores auriculares inalámbricos" -n 10 -o auriculares.json

requirements.txt sugerido:
    requests>=2.32
    beautifulsoup4>=4.12
    serpapi>=0.2
    tqdm>=4.68
"""

import os
import sys
import json
import time
import logging
import argparse
from pathlib import Path
from typing import List, Dict, Any, Tuple

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from bs4 import BeautifulSoup
from serpapi import GoogleSearch

__version__ = "1.0.0"


def create_session(user_agent: str,
                   retries: int = 3,
                   backoff_factor: float = 1.0) -> requests.Session:
    """Crea una sesión de requests con reintentos exponenciales.

    Args:
        user_agent: cabecera User-Agent a usar.
        retries: número máximo de reintentos.
        backoff_factor: factor de backoff exponencial.

    Returns:
        Session configurada.
    """
    session = requests.Session()
    retry = Retry(
        total=retries,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    session.headers.update({"User-Agent": user_agent})
    return session


def fetch_url(url: str,
              session: requests.Session,
              timeout: float,
              logger: logging.Logger
              ) -> Tuple[str, str]:
    """Descarga el contenido HTML de una URL.

    Args:
        url: página a descargar.
        session: sesión de requests configurada.
        timeout: timeout de la petición.
        logger: objeto Logger para registrar errores.

    Returns:
        Tupla (url, html) o (url, "") si falla.
    """
    try:
        resp = session.get(url, timeout=timeout)
        resp.raise_for_status()
        return url, resp.text
    except requests.RequestException as e:
        logger.error("Error descargando %s: %s", url, e)
        return url, ""


def extract_content(html: str) -> Dict[str, Any]:
    """Extrae title, h2 y párrafos principales de un HTML.

    Args:
        html: contenido HTML de la página.

    Returns:
        Diccionario con keys 'title', 'h2' y 'text'.

    >>> sample = "<h1>Título</h1><h2>Sub</h2><p>Corto</p><p>Este párrafo tiene más de cuarenta caracteres para prueba.</p>"
    >>> result = extract_content(sample)
    >>> result["title"]
    'Título'
    >>> result["h2"]
    ['Sub']
    >>> result["text"]  # comprobamos que filtra por longitud > 40
    ['Este párrafo tiene más de cuarenta caracteres para prueba.']
    """
    soup = BeautifulSoup(html, "html.parser")
    title_tag = soup.find("h1")
    title = title_tag.get_text(strip=True) if title_tag else ""
    h2_tags = [h.get_text(strip=True) for h in soup.find_all("h2")]
    paras = []
    for p in soup.find_all("p"):
        text = p.get_text(strip=True)
        if len(text) > 40:
            paras.append(text)
    return {"title": title, "h2": h2_tags, "text": paras}


def run_scraper(query: str,
                num_results: int,
                outfile: Path,
                concurrent: bool,
                user_agent: str,
                timeout: float,
                logger: logging.Logger) -> None:
    """Función principal que orquesta la búsqueda, descarga y extracción."""
    api_key = os.getenv("SERPAPI_KEY")
    if not api_key:
        logger.error("La variable de entorno SERPAPI_KEY no está definida.")
        sys.exit(1)

    params = {
        "q": query,
        "api_key": api_key,
        "num": num_results,
    }
    logger.info("Buscando '%s' en SerpAPI (n=%d)...", query, num_results)
    search = GoogleSearch(params)
    try:
        data = search.get_dict()
    except Exception as e:
        logger.error("Error con SerpAPI: %s", e)
        sys.exit(1)

    organic = data.get("organic_results", [])
    urls = [r.get("link") or r.get("url") for r in organic][:num_results]
    logger.debug("URLs obtenidas: %s", urls)

    session = create_session(user_agent)
    results: List[Dict[str, Any]] = []

    if concurrent:
        from concurrent.futures import ThreadPoolExecutor, as_completed
        logger.info("Descarga concurrente activada.")
        with ThreadPoolExecutor(max_workers=min(32, num_results)) as executor:
            futures = {
                executor.submit(fetch_url, url, session, timeout, logger): url
                for url in urls
            }
            for future in as_completed(futures):
                url, html = future.result()
                content = extract_content(html) if html else {"title": "", "h2": [], "text": []}
                results.append({"url": url, **content})
    else:
        logger.info("Descarga secuencial.")
        for url in urls:
            _, html = fetch_url(url, session, timeout, logger)
            content = extract_content(html) if html else {"title": "", "h2": [], "text": []}
            results.append({"url": url, **content})

    # Guardar JSON
    outfile.parent.mkdir(parents=True, exist_ok=True)
    with outfile.open("w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    logger.info("Resultados guardados en %s", outfile)


def parse_args() -> argparse.Namespace:
    """Parsea los argumentos de línea de comandos."""
    parser = argparse.ArgumentParser(description="SEO Scraper con SerpAPI")
    parser.add_argument("query", help="Cadena de búsqueda a consultar")
    parser.add_argument("-n", "--results", type=int, default=10,
                        choices=range(1, 51), metavar="[1-50]",
                        help="Número de resultados (1-50)")
    parser.add_argument("-o", "--outfile", type=Path, default=Path("results.json"),
                        help="Ruta del JSON de salida")
    parser.add_argument("--concurrent/--no-concurrent", dest="concurrent",
                        action="store_true", help="Activar descarga paralela")
    parser.set_defaults(concurrent=False)
    parser.add_argument("-v", "--verbose", action="count", default=0,
                        help="Nivel de verbosidad: -v INFO, -vv DEBUG")
    parser.add_argument("--version", action="version",
                        version=f"%(prog)s {__version__}")
    return parser.parse_args()


def configure_logging(verbosity: int) -> None:
    """Configura el logging según el nivel de verbosidad."""
    level = logging.WARNING
    if verbosity >= 2:
        level = logging.DEBUG
    elif verbosity == 1:
        level = logging.INFO
    logging.basicConfig(
        level=level,
        format="[%(asctime)s] %(levelname)s:%(name)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )


if __name__ == "__main__":
    args = parse_args()
    configure_logging(args.verbose)
    log = logging.getLogger("seo_scraper")

    # Obtener variables de entorno
    ua = os.getenv("REQUESTS_UA", f"seo-scraper/{__version__} (+https://example.com)")
    try:
        timeout_val = float(os.getenv("HTTP_TIMEOUT", "10"))
    except ValueError:
        log.warning("HTTP_TIMEOUT no es un número válido, usando 10s por defecto.")
        timeout_val = 10.0

    run_scraper(
        query=args.query,
        num_results=args.results,
        outfile=args.outfile,
        concurrent=args.concurrent,
        user_agent=ua,
        timeout=timeout_val,
        logger=log,
    )

    # Ejecutar tests de doctest mínimos
    import doctest
    doctest.testmod()

Funcionamiento y uso

  1. Dependencias: pip install google-search-results beautifulsoup4 requests.
  2. Clave: Regístrate en SerpAPI —100 consultas/mes gratuitas; 5 000 búsquedas cuestan 75 USD/mes.
  3. Ejecución: define la variable SERPAPI_KEY y corre el script; obtendrás ≈ 2 KB por página en JSON, listo para análisis.

3. Prompt 2 – Scraping directo con requests + BeautifulSoup

Otra manera de hacerlo es scrapeando directamente las SERP, pero te advierto que es complejo y no suele funcionar a la primera. Google se defiende del scrapeo.

### Rol asignado a la IA

Eres un **Ingeniero/a de Datos SEO sénior** con amplia experiencia en Python (≥ 3.9), que sigue las guías PEP 8 + PEP 257, aplica tipado estático, y optimiza tanto rendimiento como robustez de código.

---

### Objetivo general del script (un solo `google_scraper.py`)

1. **Buscar en Google** sin emplear APIs externas (únicamente peticiones `requests` a `https://www.google.com/search`).
2. **Paginar** hasta obtener un máximo de *n* (1-30, por defecto 10) resultados orgánicos.
3. **Extraer** la URL de cada resultado orgánico (ignorar anuncios, vídeos, mapas, etc.).
4. **Visitar** cada URL respetando:

   * **Rotación básica de *User-Agent*** (lista interna o leída de un `.txt`).
   * **Delay pseudo-aleatorio 2–5 s** entre peticiones para reducir huella.
   * **Timeout** configurable y reintentos exponenciales (back-off).
5. **Parsear** con `BeautifulSoup` cada página destino y obtener:

   ```text
   - Primer <h1>
   - Todos los <h2>
   - Párrafos de “texto principal”  (longitud > 40 caracteres, no vacíos)
   ```
6. **Guardar** un único fichero JSON UTF-8 con la forma:

   ```json
   [
     {
       "url":   "https://ejemplo.com",
       "title": "Título H1",
       "h2":    ["Subtítulo 1", "Subtítulo 2"],
       "text":  ["Párrafo 1", "Párrafo 2"]
     }
   ]
   ```
7. **Mostrar un menú/CLI** claro que permita al usuario controlar la ejecución.

---

### Interfaz de línea de comandos (usa `argparse` o `click`)

| Opción                           | Descripción                                                    |
| -------------------------------- | -------------------------------------------------------------- |
| `query` *(posicional)*           | Cadena de búsqueda obligatoria.                                |
| `-n, --results`                  | Número de resultados (1-30), **defecto 10**.                   |
| `-o, --outfile`                  | Ruta del JSON de salida, **defecto results.json**.             |
| `--ua-file`                      | Ruta opcional a fichero con *user-agents* (uno por línea).     |
| `--concurrent / --no-concurrent` | Activa descarga paralela con `asyncio` o `concurrent.futures`. |
| `-v, --verbose`                  | Nivel de *logging* (INFO/DEBUG).                               |
| `--version`                      | Muestra la versión del script.                                 |

---

### Buenas prácticas exigidas

* **Tipado** exhaustivo con *type hints* y verificación opcional `mypy`.
* **Docstrings** estilo *Google* o *NumPy* para **todas** las funciones y clases.
* **Logging** (no `print`) y barra de progreso opcional con `tqdm`.
* **Estructura lógica** en funciones/módulos; evitar bloques monolíticos.
* **`if __name__ == "__main__":`** como punto de entrada.
* **Manejo fino de excepciones**: HTTP 4xx/5xx, errores de conexión/SSL, bloqueos *captcha*, etc.
* **Tests unitarios ligeros** (`pytest` o doctest) al final del archivo dentro del bloque main.

---

### Variables de entorno aceptadas

| Variable       | Propósito                                           | Valor por defecto                       |
| -------------- | --------------------------------------------------- | --------------------------------------- |
| `REQUESTS_UA`  | *User-Agent* por defecto si no se pasa `--ua-file`. | Uno genérico para navegadores modernos. |
| `HTTP_TIMEOUT` | Timeout global en segundos.                         | `10`                                    |

---

### Instalación y ejecución (añadir en el docstring principal)

```bash
# 1. Crear entorno virtual
python -m venv .venv && source .venv/bin/activate      # Linux/macOS
REM  o bien .venv\Scripts\activate                     # Windows

# 2. Instalar dependencias
pip install -r requirements.txt

# 3. Ejecutar
python google_scraper.py "mejores auriculares inalámbricos" -n 20 -o auriculares.json --concurrent -v
```

---

### Requisitos mínimos (`requirements.txt`)

```
requests>=2.32
beautifulsoup4>=4.12
lxml>=5.2            # parser rápido y tolerante
tqdm>=4.68
fake-useragent>=1.5  # opcional para extraer UA dinámicos
```

*(si no deseas dependencia externa para UA, genera una lista estática interna y retira esa línea).*

---

### Lineamientos extra de calidad

* Formatea el código con `ruff` o `black`.
* Comentarios solo cuando aporten contexto real.
* Cumple totalmente PEP 8 y evita *code smells* (métodos largos, variables confusas, etc.).
* La aplicación debe ejecutarse sin errores con el comando de ejemplo y producir un JSON válido que respete la estructura indicada.

---

### Criterios de aceptación

1. **Autocontenida**: todo en un único `.py`.
2. **Funcional**: devuelve resultados orgánicos reales, no anuncios.
3. **Robusta**: maneja timeouts, bloqueos y códigos de error sin abortar.
4. **Eficiente**: opción de descarga concurrente sin bloquear la IU.
5. **Documentada**: docstring inicial con instrucciones de uso y secciones bien delimitadas.
6. **Testable**: incluye al menos 3 pruebas unitarias sencillas.
7. **Estilo**: cumple normas de formato y tipado.

---

> **Entrega esperada**: un archivo `google_scraper.py` listo para ejecutar, acompañado de `requirements.txt`.

Código resultante

Un código funcional que te puede servir de base:

#!/usr/bin/env python3
"""
google_scraper.py

A self-contained Google organic results scraper.

Installation:
    # 1. Create and activate virtual environment
    python -m venv .venv && source .venv/bin/activate      # Linux/macOS
    REM  or .venv\Scripts\activate                         # Windows

    # 2. Install dependencies
    pip install -r requirements.txt

Usage:
    python google_scraper.py "mejores auriculares inalámbricos" \
        -n 20 -o auriculares.json --concurrent -v

Features:
    - No external APIs: uses requests on https://www.google.com/search
    - Pagination up to N organic results (default 10)
    - Extracts only organic URLs (ignores ads, vídeos, mapas...)
    - Rotating User-Agent (static list or via --ua-file)
    - Random 2–5 s delay between page visits
    - Configurable HTTP timeout (env HTTP_TIMEOUT, default 10 s)
    - Exponential backoff retries on failures
    - Parses each page with BeautifulSoup:
        * Primer <h1>
        * Todos los <h2>
        * Párrafos > 40 caracteres
    - Saves output as UTF-8 JSON list of dicts:
      [
        {
          "url": "...",
          "title": "...",
          "h2": [...],
          "text": [...]
        },
        ...
      ]
    - CLI via argparse: query, -n/--results, -o/--outfile,
      --ua-file, --concurrent/--no-concurrent, -v/--verbose, --version
    - Type hints, Google-style docstrings, logging, tqdm, robust error handling
    - Includes basic pytest unit tests at end of file
"""

import os
import sys
import time
import random
import argparse
import logging
import json
from typing import List, Dict, Any, Optional
from urllib.parse import quote_plus

import requests
from bs4 import BeautifulSoup
from tqdm import tqdm
from requests.exceptions import RequestException

__version__ = "1.0.0"

# Static list of User-Agents if no --ua-file is provided
DEFAULT_UAS: List[str] = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 13_4) AppleWebKit/605.1.15 "
    "(KHTML, like Gecko) Version/16.5 Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
]


def get_user_agents(ua_file: Optional[str] = None) -> List[str]:
    """Load list of User-Agent strings.

    Reads from `ua_file` if provided and valid; otherwise returns a default list.

    Args:
        ua_file: Path to a text file with one User-Agent per line.

    Returns:
        A non-empty list of User-Agent strings.

    Examples:
        >>> len(get_user_agents(None)) > 0
        True
    """
    if ua_file and os.path.isfile(ua_file):
        try:
            with open(ua_file, 'r', encoding='utf-8') as f:
                uas = [line.strip() for line in f if line.strip()]
                if uas:
                    return uas
        except OSError:
            logging.warning("No se pudo leer %s, usando UA por defecto", ua_file)
    return DEFAULT_UAS.copy()


def fetch_search_results(query: str,
                         num_results: int,
                         ua_list: List[str],
                         timeout: int) -> List[str]:
    """Fetch organic result URLs from Google Search.

    Args:
        query: Search query string.
        num_results: Number of organic results to retrieve (1–30).
        ua_list: List of User-Agent strings to rotate.
        timeout: HTTP timeout in seconds.

    Returns:
        List of result URLs (de-duplicated, up to num_results).
    """
    urls: List[str] = []
    session = requests.Session()
    per_page = 10
    for start in range(0, num_results, per_page):
        params = {'q': query, 'start': start}
        headers = {'User-Agent': random.choice(ua_list)}
        try:
            resp = session.get(
                "https://www.google.com/search",
                params=params,
                headers=headers,
                timeout=timeout
            )
            resp.raise_for_status()
        except RequestException as e:
            logging.error("Error fetching search page %d: %s", start // per_page + 1, e)
            break

        soup = BeautifulSoup(resp.text, 'lxml')
        # Google organic results are in div.g with a child <a><h3>
        for g in soup.select('div.g'):
            a_tag = g.find('a')
            if not a_tag or not a_tag.find('h3'):
                continue
            href = a_tag.get('href')
            if href and href.startswith('http'):
                if href not in urls:
                    urls.append(href)
            if len(urls) >= num_results:
                break
        if len(urls) >= num_results:
            break
        time.sleep(random.uniform(1.0, 2.0))  # polite pagination delay

    return urls[:num_results]


def fetch_url(url: str,
              ua_list: List[str],
              timeout: int,
              max_retries: int = 3,
              backoff_factor: float = 1.0) -> Optional[str]:
    """Fetch a single URL with retries, random delay, and UA rotation.

    Args:
        url: URL to fetch.
        ua_list: List of User-Agent strings.
        timeout: HTTP timeout in seconds.
        max_retries: Number of retry attempts on failure.
        backoff_factor: Base backoff factor in seconds.

    Returns:
        The response text if successful; otherwise None.
    """
    session = requests.Session()
    for attempt in range(1, max_retries + 1):
        headers = {'User-Agent': random.choice(ua_list)}
        delay = random.uniform(2.0, 5.0)
        logging.debug("Sleeping %.2fs before fetching %s", delay, url)
        time.sleep(delay)
        try:
            resp = session.get(url, headers=headers, timeout=timeout)
            resp.raise_for_status()
            return resp.text
        except RequestException as e:
            wait = backoff_factor * (2 ** (attempt - 1))
            logging.warning("Error fetching %s (attempt %d/%d): %s; retrying in %.1fs",
                            url, attempt, max_retries, e, wait)
            time.sleep(wait)
    logging.error("Failed to fetch %s after %d attempts", url, max_retries)
    return None


def parse_page(url: str, html: str) -> Dict[str, Any]:
    """Parse a page HTML to extract <h1>, all <h2>, and main text paragraphs.

    Args:
        url: URL of the page (for reference).
        html: HTML content of the page.

    Returns:
        A dict with keys 'url', 'title', 'h2', 'text'.
    """
    soup = BeautifulSoup(html, 'lxml')
    h1_tag = soup.find('h1')
    title = h1_tag.get_text(strip=True) if h1_tag else ''
    h2_list = [h.get_text(strip=True) for h in soup.find_all('h2')]
    paragraphs = [
        p.get_text(strip=True)
        for p in soup.find_all('p')
        if len(p.get_text(strip=True)) > 40
    ]
    return {"url": url, "title": title, "h2": h2_list, "text": paragraphs}


def scrape(query: str,
           num_results: int,
           ua_file: Optional[str],
           concurrent: bool,
           timeout: int) -> List[Dict[str, Any]]:
    """Orchestrate search and page scraping.

    Args:
        query: Search query.
        num_results: Number of results to scrape.
        ua_file: Path to UA file.
        concurrent: Whether to fetch pages concurrently.
        timeout: HTTP timeout in seconds.

    Returns:
        List of parsed page data dicts.
    """
    ua_list = get_user_agents(ua_file)
    urls = fetch_search_results(query, num_results, ua_list, timeout)
    logging.info("Found %d URLs to scrape", len(urls))

    results: List[Dict[str, Any]] = []
    if concurrent:
        from concurrent.futures import ThreadPoolExecutor
        with ThreadPoolExecutor() as executor:
            futures = {
                executor.submit(fetch_url, url, ua_list, timeout): url
                for url in urls
            }
            for future in tqdm(
                    futures,
                    total=len(futures),
                    desc="Scraping pages"
            ):
                html = future.result()
                if html:
                    results.append(parse_page(futures[future], html))
    else:
        for url in tqdm(urls, desc="Scraping pages"):
            html = fetch_url(url, ua_list, timeout)
            if html:
                results.append(parse_page(url, html))

    return results


def main() -> None:
    """Entry point for CLI."""
    parser = argparse.ArgumentParser(description="Google organic scraper")
    parser.add_argument("query", help="Cadena de búsqueda")
    parser.add_argument("-n", "--results", type=int, default=10,
                        help="Número de resultados (1-30)")
    parser.add_argument("-o", "--outfile", default="results.json",
                        help="Ruta al JSON de salida")
    parser.add_argument("--ua-file", help="Fichero con User-Agent por línea")
    parser.add_argument("--concurrent", dest="concurrent",
                        action="store_true", help="Descarga en paralelo")
    parser.add_argument("--no-concurrent", dest="concurrent",
                        action="store_false", help="Descarga secuencial")
    parser.set_defaults(concurrent=False)
    parser.add_argument("-v", "--verbose",
                        action="store_const", dest="loglevel",
                        const=logging.DEBUG, default=logging.INFO,
                        help="Nivel de logging DEBUG")
    parser.add_argument("--version", action="version",
                        version=__version__)
    args = parser.parse_args()

    if not (1 <= args.results <= 30):
        parser.error("El número de resultados debe estar entre 1 y 30.")

    logging.basicConfig(
        level=args.loglevel,
        format="%(asctime)s [%(levelname)s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    timeout = int(os.getenv("HTTP_TIMEOUT", "10"))
    data = scrape(args.query, args.results, args.ua_file,
                  args.concurrent, timeout)

    try:
        with open(args.outfile, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        logging.info("Guardados %d resultados en %s", len(data), args.outfile)
    except OSError as e:
        logging.error("No se pudo escribir el archivo %s: %s", args.outfile, e)
        sys.exit(1)


if __name__ == "__main__":
    main()


# -------------------------
# pytest unit tests
# -------------------------
def test_get_user_agents_default() -> None:
    """Test that default UA list is returned when file is missing."""
    uas = get_user_agents("nonexistent_file.txt")
    assert isinstance(uas, list) and len(uas) >= 1


def test_parse_page_basic() -> None:
    """Test parse_page with minimal HTML."""
    html = """
    <html><body>
      <h1>Main Title</h1>
      <h2>Subtitle</h2>
      <p>This paragraph is definitely longer than forty characters long.</p>
      <p>Short one.</p>
    </body></html>
    """
    result = parse_page("http://test", html)
    assert result["url"] == "http://test"
    assert result["title"] == "Main Title"
    assert result["h2"] == ["Subtitle"]
    assert len(result["text"]) == 1
    assert "longer than forty characters" in result["text"][0]


def test_fetch_search_results_mock(monkeypatch) -> None:
    """Mock Google result page to test URL extraction."""
    sample_html = '''
    <html><body>
      <div class="g"><a href="http://a.com"><h3>Title A</h3></a></div>
      <div class="g"><a href="http://b.com"><h3>Title B</h3></a></div>
      <div class="g"><a><h3>No Link</h3></a></div>
    </body></html>
    '''
    class DummyResp:
        status_code = 200
        text = sample_html
        def raise_for_status(self): pass

    called = {"count": 0}

    def fake_get(url, params, headers, timeout):
        called["count"] += 1
        return DummyResp()

    monkeypatch.setattr(requests.Session, "get", fake_get)
    urls = fetch_search_results("test", 2, DEFAULT_UAS, timeout=5)
    assert urls == ["http://a.com", "http://b.com"]

4. Tabla comparativa de alternativas

Enfoque / APITipoComplejidadCoste aprox.*VentajasDesventajas
SerpAPISaaS SERP APIBaja75 USD / 5 000 búsq.Captchas resueltos, muchos motoresPrecio elevado al escalar
Google Custom Search JSONAPI oficialMedia5 USD / 1 000 consultasEstable y legalLímite anunciado; deprecación en 2025
Apify Google Search ScraperSaaS (pay-per-result)Baja3.50 USD / 1 000 resultadosPago granular; 5 USD de créditos gratisRendimiento sujeto a cola
Scrapy + proxiesFramework PythonAltaProxy desde 0.9 USD/IPGran control, auto-throttle, pipelinesCurva de aprendizaje
Playwright headlessNavegador realMedia-AltaGratis (libre) + proxyRenderiza JS, evita detecciónRequiere recursos y tuning
Bright Data SERP APISaaSBaja1.05 USD / 1 000 reqMulti-motor, desbloqueoCoste variable; contrato

*Precios consultados en mayo 2025; pueden variar según plan o volumen.

Aprende más sobre SEO Semántico

Daniel Pajuelo
Daniel Pajuelo es ingeniero informático y SEO Senior, actualmente trabajando en Guruwalk. En su blog personal escribe sobre Inteligencia Artificial, SEO, Vibe Coding, Blockchain... Ver más

Continua leyendo

Leer más sobre: SEO, Programación