Cómo construir análisis de búsqueda inteligente con Scrapeless y Google Gemini

# Crear entorno virtual
python -m venv scrapeless-gemini-env

# Activar el entorno
scrapeless-gemini-env\Scripts\activate

# Crear entorno virtual
python3 -m venv scrapeless-gemini-env

# Activar el entorno
source scrapeless-gemini-env/bin/activate

# Crear entorno virtual
python3 -m venv scrapeless-gemini-env

# Activar el entorno
source scrapeless-gemini-env/bin/activate

pip install requests google-generativeai python-dotenv pandas

SCRAPELESS_API_TOKEN=tu_token_aquí
GEMINI_API_KEY=tu_clave_gemini_aquí

import json
import requests
import os
from dotenv import load_dotenv

load_dotenv()

class ScrapelessClient:
    def __init__(self):
        self.token = os.getenv('SCRAPELESS_API_TOKEN')
        self.host = "api.scrapeless.com"
        self.url = f"https://{self.host}/api/v1/scraper/request"
        self.headers = {"x-api-token": self.token}
    
    def search_google(self, query, **kwargs):
        """Realizar una búsqueda en Google usando Scrapeless"""
        payload = {
            "actor": "scraper.google.search",
            "input": {
                "q": query,
                "gl": kwargs.get("gl", "us"),
                "hl": kwargs.get("hl", "en"),
                "google_domain": kwargs.get("google_domain", "google.com"),
                "location": kwargs.get("location", ""),
                "tbs": kwargs.get("tbs", ""),
                "start": str(kwargs.get("start", 0)),
                "num": str(kwargs.get("num", 10))
            }
        }
        
        response = requests.post(
            self.url, 
            headers=self.headers, 
            data=json.dumps(payload)
        )
        
        if response.status_code != 200:
            print(f"Error: {response.status_code} - {response.text}")
            return None
            
        return response.json()

import google.generativeai as genai

class SearchAnalyzer:
    def __init__(self):
        genai.configure(api_key=os.getenv('GEMINI_API_KEY'))
        self.model = genai.GenerativeModel('gemini-1.5-flash')
        self.scraper = ScrapelessClient()
    
    def analyze_topic(self, topic):
        """Buscar y analizar un tema"""
        # Paso 1: Obtener resultados de búsqueda
        print(f"Buscando: {topic}")
        search_results = self.scraper.search_google(topic, num=20)
        
        if not search_results:
            return None
        
        # Paso 2: Extraer información clave
        extracted_data = self._extract_results(search_results)
        
        # Paso 3: Analizar con Gemini
        prompt = f"""
        Analiza estos resultados de búsqueda sobre "{topic}":
        
        {json.dumps(extracted_data, indent=2)}
        
        Proporciona:
        1. Temas y tendencias clave
        2. Principales fuentes de información
        3. Perspectivas notables
        4. Acciones recomendadas
        
        Formatea tu respuesta como un informe claro y accionable.
        """
        
        response = self.model.generate_content(prompt)
        
        return {
            "tema": topic,
            "datos_de_búsqueda": extracted_data,
            "análisis": response.text
        }
```python
def _extraer_resultados(self, datos_busqueda):
        """Extraer datos relevantes de los resultados de búsqueda"""
        resultados = []
        
        if "resultados_orgánicos" in datos_busqueda:
            for item in datos_busqueda["resultados_orgánicos"][:10]:
                resultados.append({
                    "título": item.get("título", ""),
                    "extracto": item.get("extracto", ""),
                    "enlace": item.get("enlace", ""),
                    "fuente": item.get("enlace_mostrado", "")
                })
        
        return resultados

class MonitorCompetidores:
    def __init__(self):
        self.análisis = AnalizadorDeBúsqueda()
    
    def analizar_competidores(self, nombre_empresa, competidores):
        """Analizar una empresa con respecto a sus competidores"""
        todos_datos = {}
        
        # Buscar cada empresa
        for comp in [nombre_empresa] + competidores:
            print(f"\nAnalizando: {comp}")
            
            # Buscar noticias y actualizaciones recientes
            consulta_noticias = f"{comp} últimas noticias actualizaciones 2025"
            datos = self.análisis.analizar_tema(consulta_noticias)
            
            if datos:
                todos_datos[comp] = datos
        
        # Generar análisis comparativo
        aviso_comparativo = f"""
        Compara {nombre_empresa} con los competidores basándote en estos datos:
        
        {json.dumps(todos_datos, indent=2)}
        
        Proporciona:
        1. Posicionamiento competitivo
        2. Fortalezas únicas de cada empresa
        3. Oportunidades de mercado
        4. Recomendaciones estratégicas para {nombre_empresa}
        """
        
        respuesta = self.análisis.modelo.generar_contenido(aviso_comparativo)
        
        return {
            "empresa": nombre_empresa,
            "competidores": competidores,
            "análisis": respuesta.text,
            "datos_crudos": todos_datos
        }

class MonitorTendencias:
    def __init__(self):
        self.análisis = AnalizadorDeBúsqueda()
        self.raspador = ClienteSinRaspado()
    
    def monitorear_tendencia(self, palabra_clave, rango_tiempo="d"):
        """Monitorear tendencias para una palabra clave"""
        # Mapear rangos de tiempo al parámetro tbs de Google
        mapa_tiempo = {
            "h": "qdr:h",  # Última hora
            "d": "qdr:d",  # Último día
            "w": "qdr:w",  # Última semana
            "m": "qdr:m"   # Último mes
        }
        
        # Buscar con filtro de tiempo
        resultados = self.raspador.buscar_google(
            palabra_clave, 
            tbs=mapa_tiempo.get(rango_tiempo, "qdr:d"),
            num=30
        )
        
        if not resultados:
            return None
        
        # Analizar tendencias
        aviso = f"""
        Analiza las tendencias de "{palabra_clave}" basándote en los resultados de búsqueda recientes:
        
        {json.dumps(resultados.get("resultados_orgánicos", [])[:15], indent=2)}
        
        Identifica:
        1. Patrones emergentes
        2. Desarrollos clave
        3. Sentimiento (positivo/negativo/neutral)
        4. Predicciones futuras
        5. Insights accionables
        """
        
        respuesta = self.análisis.modelo.generar_contenido(aviso)
        
        return {
            "palabra_clave": palabra_clave,
            "rango_tiempo": rango_tiempo,
            "análisis": respuesta.text,
            "conteo_resultados": len(resultados.get("resultados_orgánicos", []))
        }

def main():
    # Inicializar componentes
    monitor_competidores = MonitorCompetidores()
    monitor_tendencias = MonitorTendencias()
    
    # Ejemplo 1: Análisis de Competidores
    # IMPORTANTE: Reemplaza estas empresas de ejemplo por tu empresa real y competidores
    print("=== ANÁLISIS DE COMPETIDORES ===")
    análisis = monitor_competidores.analizar_competidores(
        nombre_empresa="OpenAI",  # Reemplaza con el nombre de tu empresa
        competidores=["Anthropic", "Google AI", "Meta AI"]  # Reemplaza con tus competidores
    )
    
    print("\nResultados del Análisis Competitivo:")
    print(análisis["análisis"])
    
    # Guardar resultados
    with open("análisis_competidores.json", "w") as f:
        json.dump(análisis, f, indent=2)
    
    # Ejemplo 2: Monitoreo de Tendencias
    # IMPORTANTE: Reemplaza la palabra clave por tus términos específicos de la industria
    print("\n\n=== MONITOREO DE TENDENCIAS ===")
    tendencias = monitor_tendencias.monitorear_tendencia(
        palabra_clave="regulación de inteligencia artificial",  # Reemplaza con tus palabras clave
        rango_tiempo="w"  # Última semana
    )
    
    print("\nResultados del Análisis de Tendencias:")
    print(tendencias["análisis"])
    
    # Ejemplo 3: Análisis de Múltiples Temas
    # IMPORTANTE: Reemplaza estos temas por tus temas específicos de la industria
    print("\n\n=== ANÁLISIS DE MÚLTIPLES TEMAS ===")
    temas = [
        "aplicaciones empresariales de IA generativa",  # Reemplaza con tus temas
        "amenazas cibernéticas de IA",

    # Paso 2: Extraer información clave
    datos_extraídos = self._extract_results(search_results)
    
    # Paso 3: Analizar con Gemini
    aviso = f"""
    Analiza estos resultados de búsqueda sobre "{tema}":
    
    {json.dumps(datos_extraídos, indent=2)}
    
    Proporciona:
    1. Temas y tendencias clave
    2. Principales fuentes de información
    3. Perspectivas notables
    4. Acciones recomendadas
    
    Formatea tu respuesta como un informe claro y accionable.
    """
    
    respuesta = self.model.generate_content(aviso)
    
    return {
        "tema": tema,
        "datos_de_búsqueda": datos_extraídos,
        "análisis": respuesta.text
    }

def _extract_results(self, datos_de_búsqueda):
    """Extraer datos relevantes de los resultados de búsqueda"""
    resultados = []
    
    if "organic_results" in datos_de_búsqueda:
        for item in datos_de_búsqueda["organic_results"][:10]:
            resultados.append({
                "título": item.get("title", ""),
                "fragmento": item.get("snippet", ""),
                "enlace": item.get("link", ""),
                "fuente": item.get("displayed_link", "")
            })
    
    return resultados

def analyze_competitors(self, nombre_empresa, competidores):
    """Analiza una empresa frente a sus competidores"""
    todos_los_datos = {}
    
    # Buscar para cada empresa
    for comp in [nombre_empresa] + competidores:
        print(f"\nAnalizando: {comp}")
        
        # Buscar noticias y actualizaciones recientes
        consulta_noticias = f"{comp} últimas noticias actualizaciones 2025"
        datos = self.analyzer.analyze_topic(consulta_noticias)
        
        if datos:
            todos_los_datos[comp] = datos
        
        # Añadir un retraso entre solicitudes
        time.sleep(2)
    
    # Generar análisis comparativo
    aviso_comparativo = f"""
    Compara {nombre_empresa} con competidores basado en estos datos:
    
    {json.dumps(todos_los_datos, indent=2)}
    
    Proporciona:
    1. Posicionamiento competitivo
    2. Fortalezas únicas de cada empresa
    3. Oportunidades de mercado
    4. Recomendaciones estratégicas para {nombre_empresa}
    """
    
    respuesta = self.analyzer.model.generate_content(aviso_comparativo)
    
    return {
        "empresa": nombre_empresa,
        "competidores": competidores,
        "análisis": respuesta.text,
        "datos_raw": todos_los_datos
    }

def monitor_trend(self, palabra_clave, rango_tiempo="d"):
    """Monitorear tendencias para una palabra clave"""
    # Mapear rangos de tiempo al parámetro tbs de Google
    mapa_tiempo = {
        "h": "qdr:h",  # Última hora
        "d": "qdr:d",  # Último día
        "w": "qdr:w",  # Última semana
        "m": "qdr:m"   # Último mes
    }
    
    # Buscar con filtro de tiempo
    resultados = self.scraper.search_google(
        palabra_clave, 
        tbs=mapa_tiempo.get(rango_tiempo, "qdr:d"),
        num=30
    )
    
    if not resultados:
        return None
    
    # Analizar tendencias
    aviso = f"""
    Analiza las tendencias para "{palabra_clave}" basado en resultados de búsqueda recientes:
    
    {json.dumps(resultados.get("organic_results", [])[:15], indent=2)}
    
    Identifica:
    1. Patrones emergentes
    2. Desarrollos clave
    3. Sentimiento (positivo/negativo/neutro)
    4. Predicciones futuras
    5. Perspectivas accionables
    """
    
    respuesta = self.analyzer.model.generate_content(aviso)
    
    return {
        "palabra_clave": palabra_clave,
        "rango_tiempo": rango_tiempo,
        "análisis": respuesta.text,
        "conteo_resultados": len(resultados.get("organic_results", []))
    }

    # Aplanar la estructura de datos
    filas = []
    for item in datos:
        if isinstance(item, dict) and "search_data" in item:
            for resultado in item["search_data"]:
                filas.append({
                    "tema": item.get("tema", ""),
                    "título": resultado.get("title", ""),
                    "fragmento": resultado.get("snippet", ""),
                    "enlace": resultado.get("link", ""),
                    "fuente": resultado.get("source", "")
                })
    
    df = pd.DataFrame(filas)

df.to_csv(nombre_archivo, index=False)
        print(f"Datos exportados a {nombre_archivo}")
        
        return nombre_archivo
    
    @staticmethod
    def crear_reporte_html(datos_analisis, nombre_archivo="reporte.html"):
        """Crear un reporte HTML a partir de los datos de análisis"""
        html = f"""
        <!DOCTYPE html>
        <html>
        <head>
            <title>Reporte de Análisis de Búsqueda</title>
            <style>
                body {{ font-family: Arial, sans-serif; margin: 40px; }}
                .section {{ margin-bottom: 30px; padding: 20px; 
                           background-color: #f5f5f5; border-radius: 8px; }}
                h1 {{ color: #333; }}
                h2 {{ color: #666; }}
                pre {{ white-space: pre-wrap; word-wrap: break-word; }}
            </style>
        </head>
        <body>
            <h1>Reporte de Análisis de Búsqueda</h1>
            <p>Generado: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
            
            <div class="section">
                <h2>Resultados del Análisis</h2>
                <pre>{datos_analisis.get('analysis', 'No hay análisis disponible')}</pre>
            </div>
            
            <div class="section">
                <h2>Resumen de Datos</h2>
                <p>Tema: {datos_analisis.get('topic', 'N/A')}</p>
                <p>Resultados analizados: {len(datos_analisis.get('search_data', []))}</p>
            </div>
        </body>
        </html>
        """
        
        with open(nombre_archivo, 'w') as f:
            f.write(html)
        
        print(f"Reporte HTML creado: {nombre_archivo}")

def main():
    # Inicializar componentes
    monitor_competencia = CompetitorMonitor()
    monitor_tendencias = TrendMonitor()
    exportador = DataExporter()
    
    # Ejemplo 1: Análisis de Competencia
    print("=== ANÁLISIS DE COMPETENCIA ===")
    analisis = monitor_competencia.analyze_competitors(
        company_name="OpenAI",
        competitors=["Anthropic", "Google AI", "Meta AI"]
    )
    
    print("\nResultados del Análisis Competitivo:")
    print(analisis["analysis"])
    
    # Guardar resultados
    with open("analisis_producto.json", "w") as f:
        json.dump(analisis, f, indent=2)
    
    # Ejemplo 2: Monitoreo de Tendencias
    print("\n\n=== MONITOREO DE TENDENCIAS ===")
    tendencias = monitor_tendencias.monitor_trend(
        keyword="regulación de inteligencia artificial",
        time_range="w"  # Semana pasada
    )
    
    print("\nResultados del Análisis de Tendencias:")
    print(tendencias["analysis"])
    
    # Ejemplo 3: Análisis Multitemático
    print("\n\n=== ANÁLISIS MULTITEMÁTICO ===")
    temas = [
        "aplicaciones empresariales de IA generativa",
        "amenazas cibernéticas de IA",
        "salud en aprendizaje automático"
    ]
    
    analizador = SearchAnalyzer()
    todos_los_resultados = []
    
    for tema in temas:
        print(f"\nAnalizando: {tema}")
        resultado = analizador.analyze_topic(tema)
        
        if resultado:
            todos_los_resultados.append(resultado)
            # Guardar análisis individual
            nombre_archivo = f"{tema.replace(' ', '_')}_analisis.txt"
            with open(nombre_archivo, "w") as f:
                f.write(resultado["analysis"])
            print(f"Análisis guardado en {nombre_archivo}")
        
        # Retraso entre solicitudes
        time.sleep(2)
    
    # Exportar todos los resultados a CSV
    if todos_los_resultados:
        archivo_csv = exportador.export_to_csv(todos_los_resultados)
        print(f"\nTodos los resultados exportados a: {archivo_csv}")
        
        # Crear reporte HTML para el primer resultado
        if todos_los_resultados[0]:
            exportador.create_html_report(todos_los_resultados[0])

if __name__ == "__main__":
    main()