API REST para SEO: Integraciones y Dashboards Unificados 2026

GET /api/v3/keywords
Headers:
  Authorization: Basic base64(login:password)

1. Redirect user a Google → user autoriza
2. Google devuelve code
3. Exchange code por access_token
4. Usa access_token en requests: Authorization: Bearer token

seo-api/
├── .env                    # Variables de entorno (API keys)
├── requirements.txt        # Dependencias Python (o package.json)
├── main.py                # Entry point FastAPI (o index.js)
├── config.py              # Configuración centralizada
├── utils/
│   ├── logger.py          # Logging consistente
│   ├── error_handler.py   # Errores estándar
│   └── rate_limiter.py    # Rate limiting
├── routes/
│   ├── keywords.py        # Endpoint 1
│   ├── rankings.py        # Endpoint 2
│   ├── audit.py           # Endpoint 3
│   └── report.py          # Endpoint 4
└── models/
│   ├── request.py         # Pydantic schemas (o validadores JS)
│   └── response.py        # Response schemas

# Python FastAPI
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    # DataForSEO
    DATAFORSEO_LOGIN = os.getenv('DATAFORSEO_LOGIN')
    DATAFORSEO_PASSWORD = os.getenv('DATAFORSEO_PASSWORD')
    
    # Google OAuth
    GOOGLE_CLIENT_ID = os.getenv('GOOGLE_CLIENT_ID')
    GOOGLE_CLIENT_SECRET = os.getenv('GOOGLE_CLIENT_SECRET')
    
    # Rate limiting
    MAX_REQUESTS_PER_MINUTE = int(os.getenv('MAX_REQUESTS_PER_MINUTE', '60'))
    
    # Logging
    LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
    
    # Database (opcional, para históricos)
    DATABASE_URL = os.getenv('DATABASE_URL')

config = Config()

# DataForSEO credentials
[email protected]
DATAFORSEO_PASSWORD=tu_password_aqui

# Google OAuth (crea en console.cloud.google.com)
GOOGLE_CLIENT_ID=tu_client_id.apps.googleusercontent.com
GOOGLE_CLIENT_SECRET=tu_client_secret

# Rate limiting
MAX_REQUESTS_PER_MINUTE=60

# Logging
LOG_LEVEL=INFO

# Python FastAPI
from fastapi import HTTPException
from fastapi.responses import JSONResponse

class SEOAPIException(Exception):
    """Excepción base para errores de la API SEO"""
    def __init__(self, message: str, status_code: int = 500):
        self.message = message
        self.status_code = status_code

class RateLimitError(SEOAPIException):
    pass

class ExternalAPIError(SEOAPIException):
    pass

async def seo_exception_handler(request, exc: SEOAPIException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error": exc.message,
            "endpoint": request.url.path
        }
    )

# Python FastAPI
import logging
from datetime import datetime

def setup_logger(name: str):
    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)
    
    handler = logging.StreamHandler()
    formatter = logging.Formatter(
        '%(asctime)s | %(name)s | %(levelname)s | %(message)s'
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    
    return logger

logger = setup_logger('seo-api')

{
  "seed_keyword": "neumaticos online",
  "location_code": 2724,  # Spain
  "language_code": "es",
  "limit": 50
}

{
  "success": true,
  "data": {
    "seed": "neumaticos online",
    "opportunities": [
      {
        "keyword": "comprar neumaticos online",
        "search_volume": 2400,
        "competition": 0.6,
        "cpc": 0.85,
        "opportunity_score": 85
      },
      ...
    ],
    "total": 50,
    "generated_at": "2026-02-07T10:30:00Z"
  }
}

# routes/keywords.py - Python FastAPI
from fastapi import APIRouter, HTTPException
import requests
import os
from datetime import datetime
from utils.logger import logger
from config import config

router = APIRouter(prefix="/api/keywords", tags=["keywords"])

# DataForSEO authentication
DATAFORSEO_AUTH = (config.DATAFORSEO_LOGIN, config.DATAFORSEO_PASSWORD)
DATAFORSEO_URL = "https://api.dataforseo.com/v3/dataforseo_labs/google/keyword_suggestions/live"

def calculate_opportunity_score(keyword_data):
    """
    Calcula score de oportunidad (0-100) basado en:
    - Search volume (40%)
    - Competición invertida (30%)
    - CPC (30%)
    """
    volume_score = min(keyword_data.get('search_volume', 0) / 100, 100)
    
    # Competición menor = mejor (invertir)
    competition = keyword_data.get('competition', 0)
    competition_score = (1 - competition) * 100
    
    # CPC normalizado
    cpc_score = min(keyword_data.get('cpc', 0) * 20, 100)
    
    return int(
        volume_score * 0.4 + 
        competition_score * 0.3 + 
        cpc_score * 0.3
    )

@router.post("/research")
async def keyword_research(request: dict):
    """
    Endpoint para keyword research usando DataForSEO API
    
    Input:
    - seed_keyword: Palabra semilla
    - location_code: Código de ubicación (2724 = España)
    - language_code: Código de idioma (es = español)
    - limit: Máximo de resultados
    
    Output:
    - Lista de keywords ordenadas por opportunity_score
    """
    
    # Validar input
    seed_keyword = request.get('seed_keyword')
    if not seed_keyword:
        raise HTTPException(status_code=400, detail="seed_keyword es requerido")
    
    location_code = request.get('location_code', 2724)
    language_code = request.get('language_code', 'es')
    limit = min(request.get('limit', 50), 100)  # Max 100
    
    logger.info(f"Keyword research iniciado: {seed_keyword}")
    
    # Preparar payload para DataForSEO
    payload = [{
        "keyword": seed_keyword,
        "location_code": location_code,
        "language_code": language_code,
        "limit": limit
    }]
    
    try:
        # Consultar DataForSEO API
        response = requests.post(
            DATAFORSEO_URL,
            json=payload,
            auth=DATAFORSEO_AUTH,
            timeout=30
        )
        
        if response.status_code != 200:
            logger.error(f"DataForSEO error: {response.status_code}")
            raise HTTPException(
                status_code=502,
                detail="Error al consultar DataForSEO API"
            )
        
        data = response.json()
        
        if 'tasks' not in data or not data['tasks']:
            raise HTTPException(
                status_code=500,
                detail="Respuesta inválida de DataForSEO"
            )
        
        # Extraer y procesar keywords
        raw_keywords = data['tasks'][0].get('result', [])
        
        opportunities = []
        for kw in raw_keywords:
            opp_score = calculate_opportunity_score(kw)
            
            opportunities.append({
                "keyword": kw.get('keyword'),
                "search_volume": kw.get('search_volume', 0),
                "competition": round(kw.get('competition', 0), 2),
                "cpc": round(kw.get('cpc', 0), 2),
                "opportunity_score": opp_score,
                "keyword_info": {
                    "last_update": kw.get('last_update_date')
                }
            })
        
        # Ordenar por opportunity_score descendente
        opportunities.sort(key=lambda x: x['opportunity_score'], reverse=True)
        
        logger.info(f"{len(opportunities)} keywords procesadas")
        
        return {
            "success": True,
            "data": {
                "seed": seed_keyword,
                "opportunities": opportunities,
                "total": len(opportunities),
                "generated_at": datetime.utcnow().isoformat() + "Z"
            }
        }
        
    except requests.exceptions.Timeout:
        logger.error("Timeout en DataForSEO API")
        raise HTTPException(
            status_code=504,
            detail="Timeout al consultar DataForSEO API"
        )
    except Exception as e:
        logger.error(f"Error inesperado: {str(e)}")
        raise HTTPException(
            status_code=500,
            detail=f"Error interno: {str(e)}"
        )

// routes/keywords.js - Node.js Express
const express = require('express');
const axios = require('axios');
const router = express.Router();
require('dotenv').config();

// DataForSEO authentication
const DATAFORSEO_AUTH = Buffer.from(
  `${process.env.DATAFORSEO_LOGIN}:${process.env.DATAFORSEO_PASSWORD}`
).toString('base64');

const DATAFORSEO_URL = 'https://api.dataforseo.com/v3/dataforseo_labs/google/keyword_suggestions/live';

function calculateOpportunityScore(keywordData) {
  const volumeScore = Math.min((keywordData.search_volume || 0) / 100, 100);
  const competition = keywordData.competition || 0;
  const competitionScore = (1 - competition) * 100;
  const cpcScore = Math.min((keywordData.cpc || 0) * 20, 100);
  
  return Math.floor(
    volumeScore * 0.4 + 
    competitionScore * 0.3 + 
    cpcScore * 0.3
  );
}

router.post('/research', async (req, res) => {
  try {
    const { seed_keyword, location_code = 2724, language_code = 'es', limit = 50 } = req.body;
    
    if (!seed_keyword) {
      return res.status(400).json({
        success: false,
        error: 'seed_keyword es requerido'
      });
    }
    
    console.log(`Keyword research iniciado: ${seed_keyword}`);
    
    const payload = [{
      keyword: seed_keyword,
      location_code: Number(location_code),
      language_code,
      limit: Math.min(Number(limit), 100)
    }];
    
    const response = await axios.post(DATAFORSEO_URL, payload, {
      headers: {
        'Authorization': `Basic ${DATAFORSEO_AUTH}`,
        'Content-Type': 'application/json'
      },
      timeout: 30000
    });
    
    if (response.status !== 200 || !response.data.tasks) {
      throw new Error('Respuesta inválida de DataForSEO');
    }
    
    const rawKeywords = response.data.tasks[0]?.result || [];
    
    const opportunities = rawKeywords.map(kw => ({
      keyword: kw.keyword,
      search_volume: kw.search_volume || 0,
      competition: Math.round((kw.competition || 0) * 100) / 100,
      cpc: Math.round((kw.cpc || 0) * 100) / 100,
      opportunity_score: calculateOpportunityScore(kw),
      keyword_info: {
        last_update: kw.last_update_date
      }
    }));
    
    opportunities.sort((a, b) => b.opportunity_score - a.opportunity_score);
    
    console.log(`${opportunities.length} keywords procesadas`);
    
    res.json({
      success: true,
      data: {
        seed: seed_keyword,
        opportunities,
        total: opportunities.length,
        generated_at: new Date().toISOString()
      }
    });
    
  } catch (error) {
    console.error('Error:', error.message);
    
    if (error.code === 'ECONNABORTED') {
      return res.status(504).json({
        success: false,
        error: 'Timeout al consultar DataForSEO API'
      });
    }
    
    res.status(500).json({
      success: false,
      error: error.message
    });
  }
});

module.exports = router;

# Test con curl
curl -X POST "http://localhost:8000/api/keywords/research" \
  -H "Content-Type: application/json" \
  -d '{
    "seed_keyword": "neumaticos online",
    "location_code": 2724,
    "language_code": "es",
    "limit": 20
  }'

# routes/rankings.py - Python FastAPI
from fastapi import APIRouter, Depends, BackgroundTasks
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import sqlite3
from datetime import datetime, timedelta

router = APIRouter(prefix="/api/rankings", tags=["rankings"])

# Database setup para históricos
def init_db():
    conn = sqlite3.connect('rankings.db')
    c = conn.cursor()
    c.execute('''
        CREATE TABLE IF NOT EXISTS rankings_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            keyword TEXT,
            url TEXT,
            position INTEGER,
            clicks INTEGER,
            impressions INTEGER,
            ctr REAL,
            date DATE
        )
    ''')
    conn.commit()
    conn.close()

def save_ranking_snapshot(keyword_data):
    """Guarda snapshot de rankings en database"""
    conn = sqlite3.connect('rankings.db')
    c = conn.cursor()
    
    today = datetime.now().strftime('%Y-%m-%d')
    
    for kw in keyword_data:
        c.execute('''
            INSERT INTO rankings_history 
            (keyword, url, position, clicks, impressions, ctr, date)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            kw['keyword'],
            kw['url'],
            kw['position'],
            kw['clicks'],
            kw['impressions'],
            kw['ctr'],
            today
        ))
    
    conn.commit()
    conn.close()

def detect_significant_changes(current_rankings):
    """
    Compara con snapshot anterior y detecta cambios >3 posiciones
    """
    conn = sqlite3.connect('rankings.db')
    c = conn.cursor()
    
    yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
    
    c.execute('''
        SELECT keyword, position FROM rankings_history 
        WHERE date = ?
    ''', (yesterday,))
    
    previous = {row[0]: row[1] for row in c.fetchall()}
    conn.close()
    
    changes = []
    for kw in current_rankings:
        keyword = kw['keyword']
        current_pos = kw['position']
        
        if keyword in previous:
            prev_pos = previous[keyword]
            position_change = prev_pos - current_pos
            
            if abs(position_change) >= 3:
                changes.append({
                    'keyword': keyword,
                    'previous_position': prev_pos,
                    'current_position': current_pos,
                    'change': position_change
                })
    
    return changes

@router.post("/track")
async def track_rankings(request: dict, background_tasks: BackgroundTasks):
    """
    Endpoint para tracking de rankings usando GSC API
    
    Input:
    - site_url: Propiedad GSC (ej: sc-domain:ejemplo.com)
    - keywords: Lista de keywords a trackear
    - days_back: Días de datos históricos
    
    Output:
    - Rankings actuales + cambios detectados
    """
    
    site_url = request.get('site_url')
    keywords = request.get('keywords', [])
    days_back = request.get('days_back', 7)
    
    # Autenticación con GSC OAuth (simplificado)
    credentials = Credentials.from_authorized_user_file('token.json')
    
    # Consultar GSC API (pseudo-código)
    # gsc_service = build('searchconsole', 'v1', credentials=credentials)
    # response = gsc_service.searchanalytics().query(...).execute()
    
    # Simular datos de respuesta
    current_rankings = [
        {
            'keyword': 'neumaticos online',
            'url': 'https://ejemplo.com/neumaticos',
            'position': 5.2,
            'clicks': 142,
            'impressions': 2400,
            'ctr': 5.92
        },
        # ... más keywords
    ]
    
    # Guardar snapshot en background
    background_tasks.add_task(save_ranking_snapshot, current_rankings)
    
    # Detectar cambios significativos
    changes = detect_significant_changes(current_rankings)
    
    return {
        "success": True,
        "data": {
            "current_rankings": current_rankings,
            "significant_changes": changes,
            "tracked_keywords": len(keywords),
            "generated_at": datetime.utcnow().isoformat() + "Z"
        }
    }

// routes/rankings.js - Node.js Express
const express = require('express');
const sqlite3 = require('sqlite3').verbose();
const router = express.Router();

const db = new sqlite3.Database('./rankings.db');

// Inicializar tabla
db.serialize(() => {
  db.run(`
    CREATE TABLE IF NOT EXISTS rankings_history (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      keyword TEXT,
      url TEXT,
      position REAL,
      clicks INTEGER,
      impressions INTEGER,
      ctr REAL,
      date TEXT
    )
  `);
});

function saveRankingSnapshot(keywordData, callback) {
  const stmt = db.prepare(`
    INSERT INTO rankings_history 
    (keyword, url, position, clicks, impressions, ctr, date)
    VALUES (?, ?, ?, ?, ?, ?, ?)
  `);
  
  const today = new Date().toISOString().split('T')[0];
  
  db.serialize(() => {
    keywordData.forEach(kw => {
      stmt.run(
        kw.keyword,
        kw.url,
        kw.position,
        kw.clicks,
        kw.impressions,
        kw.ctr,
        today
      );
    });
    stmt.finalize(callback);
  });
}

router.post('/track', async (req, res) => {
  const { site_url, keywords = [], days_back = 7 } = req.body;
  
  // Consultar GSC API (pseudo-código)
  const currentRankings = [
    {
      keyword: 'neumaticos online',
      url: 'https://ejemplo.com/neumaticos',
      position: 5.2,
      clicks: 142,
      impressions: 2400,
      ctr: 5.92
    }
  ];
  
  // Guardar snapshot
  saveRankingSnapshot(currentRankings, () => {
    res.json({
      success: true,
      data: {
        current_rankings: currentRankings,
        total: currentRankings.length
      }
    });
  });
});

module.exports = router;

# routes/audit.py - Python FastAPI
from fastapi import APIRouter, HTTPException
import asyncio
import aiohttp
from urllib.parse import urljoin
from bs4 import BeautifulSoup
from datetime import datetime

router = APIRouter(prefix="/api/audit", tags=["audit"])

# Checks modulares
async def check_title(url, session):
    """Verifica que la página tenga title tag"""
    try:
        async with session.get(url, timeout=10) as response:
            html = await response.text()
            soup = BeautifulSoup(html, 'html.parser')
            title = soup.find('title')
            
            if not title or not title.string.strip():
                return {
                    'check': 'title',
                    'status': 'fail',
                    'issue': 'Missing or empty title tag',
                    'severity': 'high'
                }
            
            title_len = len(title.string.strip())
            if title_len < 30 or title_len > 60:
                return {
                    'check': 'title',
                    'status': 'warning',
                    'issue': f'Title length ({title_len}) outside optimal range (30-60)',
                    'severity': 'medium'
                }
            
            return {
                'check': 'title',
                'status': 'pass',
                'value': title.string.strip()
            }
    except Exception as e:
        return {
            'check': 'title',
            'status': 'error',
            'issue': str(e),
            'severity': 'high'
        }

async def check_meta_description(url, session):
    """Verifica meta description"""
    try:
        async with session.get(url, timeout=10) as response:
            html = await response.text()
            soup = BeautifulSoup(html, 'html.parser')
            meta_desc = soup.find('meta', attrs={'name': 'description'})
            
            if not meta_desc or not meta_desc.get('content'):
                return {
                    'check': 'meta_description',
                    'status': 'fail',
                    'issue': 'Missing meta description',
                    'severity': 'high'
                }
            
            desc_len = len(meta_desc.get('content'))
            if desc_len < 120 or desc_len > 160:
                return {
                    'check': 'meta_description',
                    'status': 'warning',
                    'issue': f'Meta description length ({desc_len}) outside optimal range (120-160)',
                    'severity': 'medium'
                }
            
            return {
                'check': 'meta_description',
                'status': 'pass'
            }
    except Exception as e:
        return {
            'check': 'meta_description',
            'status': 'error',
            'issue': str(e),
            'severity': 'high'
        }

async def check_h1_structure(url, session):
    """Verifica estructura de H1"""
    try:
        async with session.get(url, timeout=10) as response:
            html = await response.text()
            soup = BeautifulSoup(html, 'html.parser')
            h1s = soup.find_all('h1')
            
            if len(h1s) == 0:
                return {
                    'check': 'h1_structure',
                    'status': 'fail',
                    'issue': 'Missing H1 tag',
                    'severity': 'high'
                }
            
            if len(h1s) > 1:
                return {
                    'check': 'h1_structure',
                    'status': 'warning',
                    'issue': f'Multiple H1 tags found ({len(h1s)})',
                    'severity': 'medium'
                }
            
            return {
                'check': 'h1_structure',
                'status': 'pass',
                'value': h1s[0].text.strip()[:50]
            }
    except Exception as e:
        return {
            'check': 'h1_structure',
            'status': 'error',
            'issue': str(e),
            'severity': 'high'
        }

async def check_canonical(url, session):
    """Verifica canonical tag"""
    try:
        async with session.get(url, timeout=10) as response:
            html = await response.text()
            soup = BeautifulSoup(html, 'html.parser')
            canonical = soup.find('link', attrs={'rel': 'canonical'})
            
            if not canonical:
                return {
                    'check': 'canonical',
                    'status': 'warning',
                    'issue': 'Missing canonical tag',
                    'severity': 'medium'
                }
            
            canonical_url = canonical.get('href')
            if canonical_url != url:
                return {
                    'check': 'canonical',
                    'status': 'warning',
                    'issue': f'Canonical points to different URL: {canonical_url}',
                    'severity': 'low'
                }
            
            return {
                'check': 'canonical',
                'status': 'pass'
            }
    except Exception as e:
        return {
            'check': 'canonical',
            'status': 'error',
            'issue': str(e),
            'severity': 'medium'
        }

def calculate_audit_score(results):
    """
    Calcula score de audit (0-100)
    
    Pesos:
    - High severity issues: -15 puntos
    - Medium severity issues: -5 puntos
    - Low severity issues: -2 puntos
    - Errors: -20 puntos
    """
    score = 100
    
    for result in results:
        if result['status'] == 'error':
            score -= 20
        elif result['status'] == 'fail':
            if result['severity'] == 'high':
                score -= 15
            elif result['severity'] == 'medium':
                score -= 5
            else:
                score -= 2
        elif result['status'] == 'warning':
            if result['severity'] == 'high':
                score -= 10
            elif result['severity'] == 'medium':
                score -= 3
            else:
                score -= 1
    
    return max(score, 0)

@router.post("/run")
async def run_audit(request: dict):
    """
    Endpoint para ejecutar site audit en paralelo
    
    Input:
    - url: URL a auditar
    - checks: Lista de checks a ejecutar (todos por defecto)
    
    Output:
    - Resultados de cada check + score global
    """
    
    url = request.get('url')
    if not url:
        raise HTTPException(status_code=400, detail="URL es requerida")
    
    checks_to_run = request.get('checks', [
        'title', 'meta_description', 'h1_structure', 'canonical'
    ])
    
    # Mapear checks a funciones
    check_functions = {
        'title': check_title,
        'meta_description': check_meta_description,
        'h1_structure': check_h1_structure,
        'canonical': check_canonical
    }
    
    # Ejecutar checks en paralelo
    async with aiohttp.ClientSession() as session:
        tasks = [
            check_functions[check](url, session) 
            for check in checks_to_run 
            if check in check_functions
        ]
        
        results = await asyncio.gather(*tasks)
    
    # Calcular score
    score = calculate_audit_score(results)
    
    # Priorizar issues
    high_priority = [r for r in results if r.get('severity') == 'high' and r['status'] != 'pass']
    medium_priority = [r for r in results if r.get('severity') == 'medium' and r['status'] != 'pass']
    
    return {
        "success": True,
        "data": {
            "url": url,
            "score": score,
            "grade": "A" if score >= 90 else "B" if score >= 70 else "C" if score >= 50 else "F",
            "results": results,
            "summary": {
                "total_checks": len(results),
                "passed": len([r for r in results if r['status'] == 'pass']),
                "warnings": len([r for r in results if r['status'] == 'warning']),
                "failed": len([r for r in results if r['status'] == 'fail']),
                "errors": len([r for r in results if r['status'] == 'error'])
            },
            "priority_issues": {
                "high": len(high_priority),
                "medium": len(medium_priority)
            },
            "audited_at": datetime.utcnow().isoformat() + "Z"
        }
    }

// routes/audit.js - Node.js Express
const express = require('express');
const axios = require('axios');
const { JSDOM } = require('jsdom');
const router = express.Router();

async function checkTitle(url) {
  try {
    const response = await axios.get(url, { timeout: 10000 });
    const dom = new JSDOM(response.data);
    const title = dom.window.document.querySelector('title');
    
    if (!title || !title.textContent.trim()) {
      return {
        check: 'title',
        status: 'fail',
        issue: 'Missing or empty title tag',
        severity: 'high'
      };
    }
    
    return {
      check: 'title',
      status: 'pass',
      value: title.textContent.trim()
    };
  } catch (error) {
    return {
      check: 'title',
      status: 'error',
      issue: error.message,
      severity: 'high'
    };
  }
}

router.post('/run', async (req, res) => {
  const { url, checks = ['title', 'meta_description', 'h1', 'canonical'] } = req.body;
  
  if (!url) {
    return res.status(400).json({
      success: false,
      error: 'URL es requerida'
    });
  }
  
  // Ejecutar checks en paralelo
  const checkPromises = [
    checkTitle(url)
    // ... más checks
  ];
  
  const results = await Promise.all(checkPromises);
  
  // Calcular score
  let score = 100;
  results.forEach(result => {
    if (result.status === 'error') score -= 20;
    else if (result.status === 'fail') score -= 15;
    else if (result.status === 'warning') score -= 5;
  });
  
  score = Math.max(score, 0);
  
  res.json({
    success: true,
    data: {
      url,
      score,
      grade: score >= 90 ? 'A' : score >= 70 ? 'B' : score >= 50 ? 'C' : 'F',
      results,
      audited_at: new Date().toISOString()
    }
  });
});

module.exports = router;

# routes/report.py - Python FastAPI
from fastapi import APIRouter, HTTPException
import pandas as pd
import json
from datetime import datetime, timedelta
from jinja2 import Template
import csv
from io import StringIO

router = APIRouter(prefix="/api/report", tags=["report"])

def fetch_gsc_data(site_url, start_date, end_date):
    """Consulta GSC API para datos de búsqueda"""
    # Pseudo-código - conectar con GSC OAuth
    return {
        'clicks': 15420,
        'impressions': 342000,
        'ctr': 4.51,
        'average_position': 12.3
    }

def fetch_analytics_data(property_id, start_date, end_date):
    """Consulta Google Analytics API"""
    # Pseudo-código - conectar con GA4
    return {
        'sessions': 45230,
        'users': 38450,
        'pageviews': 128500,
        'bounce_rate': 42.3
    }

def fetch_keyword_data(seed_keywords):
    """Consulta DataForSEO para oportunidades"""
    # Pseudo-código
    return [
        {'keyword': 'neumaticos baratos', 'volume': 2400, 'opportunity_score': 85},
        {'keyword': 'neumaticos online', 'volume': 1800, 'opportunity_score': 78}
    ]

def generate_html_report(data):
    """Genera reporte HTML con template Jinja2"""
    template = Template('''
        <!DOCTYPE html>
        <html>
        
        <body>
            <h1>SEO Report - {{ date }}</h1>
            
            <h2>Google Search Console</h2>
            <div class="metric">
                <p><strong>Clicks:</strong> {{ gsc.clicks }}</p>
                <p><strong>Impressions:</strong> {{ gsc.impressions }}</p>
                <p><strong>CTR:</strong> {{ gsc.ctr }}%</p>
                <p><strong>Avg Position:</strong> {{ gsc.average_position }}</p>
            </div>
            
            <h2>Google Analytics</h2>
            <div class="metric">
                <p><strong>Sessions:</strong> {{ ga.sessions }}</p>
                <p><strong>Users:</strong> {{ ga.users }}</p>
                <p><strong>Pageviews:</strong> {{ ga.pageviews }}</p>
                <p><strong>Bounce Rate:</strong> {{ ga.bounce_rate }}%</p>
            </div>
            
            <h2>Keyword Opportunities</h2>
            <table>
                <tr><th>Keyword</th><th>Volume</th><th>Score</th></tr>
                {% for kw in keywords %}
                <tr>
                    <td>{{ kw.keyword }}</td>
                    <td>{{ kw.volume }}</td>
                    <td>{{ kw.opportunity_score }}</td>
                </tr>
                {% endfor %}
            </table>
        </body>
        </html>
    ''')
    
    return template.render(
        date=datetime.now().strftime('%Y-%m-%d'),
        gsc=data['gsc'],
        ga=data['analytics'],
        keywords=data['keywords'][:10]
    )

def generate_csv_report(data):
    """Genera reporte en formato CSV"""
    output = StringIO()
    writer = csv.writer(output)
    
    # Header
    writer.writerow(['Metric', 'Value'])
    
    # GSC data
    for key, value in data['gsc'].items():
        writer.writerow([f'GSC_{key}', value])
    
    # Analytics data
    for key, value in data['analytics'].items():
        writer.writerow([f'GA_{key}', value])
    
    return output.getvalue()

@router.post("/generate")
async def generate_report(request: dict):
    """
    Endpoint para generar reporte unificado
    
    Input:
    - site_url: Propiedad GSC
    - property_id: GA4 property ID
    - seed_keywords: Keywords para DataForSEO
    - format: json | html | csv (todos por defecto)
    - start_date: Fecha inicio (default: 30 días atrás)
    - end_date: Fecha fin (default: hoy)
    
    Output:
    - Reporte consolidado en formato(s) solicitado(s)
    """
    
    site_url = request.get('site_url')
    property_id = request.get('property_id')
    seed_keywords = request.get('seed_keywords', [])
    output_format = request.get('format', 'json')
    
    # Calcular rango de fechas
    end_date = datetime.now()
    start_date = end_date - timedelta(days=30)
    
    # Consultar todas las fuentes en paralelo
    gsc_data = fetch_gsc_data(site_url, start_date, end_date)
    analytics_data = fetch_analytics_data(property_id, start_date, end_date)
    keyword_data = fetch_keyword_data(seed_keywords)
    
    # Consolidar datos
    consolidated_data = {
        'gsc': gsc_data,
        'analytics': analytics_data,
        'keywords': keyword_data,
        'period': {
            'start': start_date.strftime('%Y-%m-%d'),
            'end': end_date.strftime('%Y-%m-%d')
        },
        'generated_at': datetime.utcnow().isoformat() + "Z"
    }
    
    # Generar output según formato
    if output_format == 'json':
        return {
            "success": True,
            "data": consolidated_data
        }
    
    elif output_format == 'html':
        html_content = generate_html_report(consolidated_data)
        return {
            "success": True,
            "content_type": "text/html",
            "data": html_content
        }
    
    elif output_format == 'csv':
        csv_content = generate_csv_report(consolidated_data)
        return {
            "success": True,
            "content_type": "text/csv",
            "data": csv_content
        }
    
    elif output_format == 'all':
        return {
            "success": True,
            "data": {
                "json": consolidated_data,
                "html": generate_html_report(consolidated_data),
                "csv": generate_csv_report(consolidated_data)
            }
        }

// routes/report.js - Node.js Express
const express = require('express');
const router = express.Router();

function fetchGSCData(siteUrl, startDate, endDate) {
  // Pseudo-código GSC API
  return {
    clicks: 15420,
    impressions: 342000,
    ctr: 4.51,
    average_position: 12.3
  };
}

function fetchAnalyticsData(propertyId, startDate, endDate) {
  // Pseudo-código GA4 API
  return {
    sessions: 45230,
    users: 38450,
    pageviews: 128500,
    bounce_rate: 42.3
  };
}

router.post('/generate', async (req, res) => {
  const { 
    site_url, 
    property_id, 
    seed_keywords = [], 
    format = 'json' 
  } = req.body;
  
  const endDate = new Date();
  const startDate = new Date();n  startDate.setDate(startDate.getDate() - 30);
  
  const gscData = fetchGSCData(site_url, startDate, endDate);
  const analyticsData = fetchAnalyticsData(property_id, startDate, endDate);
  
  const consolidatedData = {
    gsc: gscData,
    analytics: analyticsData,
    period: {
      start: startDate.toISOString().split('T')[0],
      end: endDate.toISOString().split('T')[0]
    },
    generated_at: new Date().toISOString()
  };
  
  if (format === 'json') {
    return res.json({
      success: true,
      data: consolidatedData
    });
  }
  
  // ... html, csv handling
});

module.exports = router;

# Dockerfile para FastAPI
FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  seo-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATAFORSEO_LOGIN=${DATAFORSEO_LOGIN}
      - DATAFORSEO_PASSWORD=${DATAFORSEO_PASSWORD}
    volumes:
      - ./logs:/app/logs
      - ./data:/app/data
    restart: unless-stopped

# crontab - ejecutar reporte semanal cada lunes a las 9am
0 9 * * 1 curl -X POST http://localhost:8000/api/report/generate \
  -H "Content-Type: application/json" \
  -d '{
    "site_url": "sc-domain:misitio.com",
    "property_id": "123456",
    "format": "html"
  }' > reports/weekly-$(date +\%Y-\%m-\%d).html