GSC + Python: lo script per monitorare keyword in posizione 6-20

setup rapido

# 1. Google Cloud Console → Nuovo progetto → API e servizi → Libreria
#    Cerca "Google Search Console API" → Abilita

# 2. Credenziali → Crea credenziali → Account di servizio
#    Scarica il file JSON → salvalo come service-account.json

# 3. Search Console → Impostazioni → Utenti e autorizzazioni
#    Aggiungi: nome@progetto.iam.gserviceaccount.com → Proprietario

# 4. Installa librerie
pip install google-auth google-auth-httplib2 google-api-python-client pandas

gsc_keyword_monitor.py

"""
GSC Keyword Monitor v1.0
Monitora keyword in posizione 6-20, calcola potenziale di clic,
confronta con periodo precedente, esporta CSV.
"""

from google.oauth2 import service_account
from googleapiclient.discovery import build
from datetime import date, timedelta
from pathlib import Path
import pandas as pd
import json
import os

# ============================================================
# CONFIGURAZIONE
# ============================================================
CONFIG = {
    'key_file':    'service-account.json',
    'site_url':    'https://smartweb-media.com/',
    'scopes':      ['https://www.googleapis.com/auth/webmasters.readonly'],
    'days':        28,           # finestra temporale
    'pos_min':     6.0,          # posizione minima quick wins
    'pos_max':     20.0,         # posizione massima quick wins
    'min_imps':    3,            # impressioni minime da includere
    'output_dir':  'gsc_reports',
    'row_limit':   5000,
}

# CTR benchmark per posizione (stime conservative)
CTR_BENCHMARK = {
    1: 0.28, 2: 0.15, 3: 0.11, 4: 0.09, 5: 0.07,
    6: 0.05, 7: 0.04, 8: 0.03, 9: 0.025, 10: 0.02,
    11: 0.018, 12: 0.015, 13: 0.013, 14: 0.012, 15: 0.010,
    16: 0.009, 17: 0.008, 18: 0.007, 19: 0.006, 20: 0.005,
}

# ============================================================
# AUTENTICAZIONE
# ============================================================
def get_service(key_file, scopes):
    """Crea il client GSC autenticato con service account."""
    creds = service_account.Credentials.from_service_account_file(
        key_file, scopes=scopes)
    return build('searchconsole', 'v1', credentials=creds)

# ============================================================
# ESTRAZIONE DATI
# ============================================================
def fetch_search_analytics(service, site_url, start_date, end_date,
                           dimensions=None, row_limit=5000):
    """Scarica dati Search Analytics dalla GSC API."""
    if dimensions is None:
        dimensions = ['query', 'page']

    response = service.searchanalytics().query(
        siteUrl=site_url,
        body={
            'startDate': str(start_date),
            'endDate': str(end_date),
            'dimensions': dimensions,
            'rowLimit': row_limit,
            'orderBy': [{'fieldName': 'impressions', 'sortOrder': 'DESCENDING'}]
        }
    ).execute()
    return response.get('rows', [])

def rows_to_dataframe(rows):
    """Converte le righe GSC in un DataFrame pandas."""
    if not rows:
        return pd.DataFrame()

    records = []
    for row in rows:
        record = {k: v for k, v in zip(['query', 'page'], row['keys'])}
        record.update({
            'clicks':      row.get('clicks', 0),
            'impressions': row.get('impressions', 0),
            'ctr':         row.get('ctr', 0),
            'position':    row.get('position', 0)
        })
        records.append(record)
    return pd.DataFrame(records)

# ============================================================
# ANALISI QUICK WINS
# ============================================================
def calculate_potential(df, pos_min, pos_max, min_imps, ctr_benchmark):
    """Identifica quick wins e calcola il potenziale di traffico."""
    # Filtra il range di posizioni
    mask = (
        (df['position'] >= pos_min) &
        (df['position'] <= pos_max) &
        (df['impressions'] >= min_imps)
    )
    wins = df[mask].copy()

    if wins.empty:
        return wins

    # Posizione target per il calcolo del gain (top 3)
    target_pos = 3
    target_ctr = ctr_benchmark.get(target_pos, 0.11)

    # Calcola gain stimato (clic aggiuntivi al mese se arriva a pos 3)
    wins['target_ctr']      = target_ctr
    wins['current_ctr_pct'] = (wins['ctr'] * 100).round(2)
    wins['gain_clic_stimato'] = (
        (target_ctr - wins['ctr']) * wins['impressions']
    ).clip(lower=0).round(0).astype(int)

    # Score opportunità: pesa impressioni + gap CTR + vicinanza top 10
    wins['proximity_score'] = (10 - wins['position'].clip(upper=10)) / 10
    wins['opp_score'] = (
        wins['gain_clic_stimato'] * 0.5 +
        wins['impressions'] * 0.3 +
        wins['proximity_score'] * 100 * 0.2
    ).round(1)

    return wins.sort_values('opp_score', ascending=False)

# ============================================================
# CONFRONTO PERIODI
# ============================================================
def compare_periods(curr_df, prev_df):
    """Confronta posizioni attuali con periodo precedente."""
    if curr_df.empty or prev_df.empty:
        return curr_df

    # Posizione media per query nel periodo corrente
    curr_pos = curr_df.groupby('query')['position'].mean().rename('pos_curr')
    prev_pos = prev_df.groupby('query')['position'].mean().rename('pos_prev')

    comparison = pd.concat([curr_pos, prev_pos], axis=1).dropna()
    comparison['delta_pos'] = (comparison['pos_prev'] - comparison['pos_curr']).round(1)
    comparison['trend'] = comparison['delta_pos'].apply(
        lambda x: '↑ Salendo' if x > 0.5 else ('↓ Scendendo' if x < -0.5 else '→ Stabile')
    )
    return comparison.reset_index()

# ============================================================
# EXPORT CSV
# ============================================================
def export_reports(quick_wins, comparison, output_dir, site_url):
    """Esporta i report in CSV con timestamp."""
    Path(output_dir).mkdir(exist_ok=True)
    today = date.today().isoformat()

    # Quick wins
    if not quick_wins.empty:
        qw_file = f"{output_dir}/quick_wins_{today}.csv"
        cols = ['query', 'page', 'position', 'impressions', 'clicks',
                'current_ctr_pct', 'gain_clic_stimato', 'opp_score']
        available = [c for c in cols if c in quick_wins.columns]
        quick_wins[available].to_csv(qw_file, index=False)
        print(f"  Quick wins salvati: {qw_file}")

    # Comparazione periodi
    if not comparison.empty:
        cmp_file = f"{output_dir}/movements_{today}.csv"
        comparison.to_csv(cmp_file, index=False)
        print(f"  Movimenti salvati:  {cmp_file}")

    # Summary JSON
    summary = {
        'date': today,
        'site': site_url,
        'quick_wins_count': len(quick_wins),
        'total_gain_potential': int(quick_wins['gain_clic_stimato'].sum()) if not quick_wins.empty else 0,
        'top_opportunities': quick_wins.head(5)[['query','position','gain_clic_stimato']].to_dict('records') if not quick_wins.empty else []
    }
    with open(f"{output_dir}/summary_{today}.json", 'w') as f:
        json.dump(summary, f, ensure_ascii=False, indent=2)

# ============================================================
# MAIN
# ============================================================
def main():
    cfg = CONFIG
    end_curr  = date.today()
    start_curr = end_curr - timedelta(days=cfg['days'])
    end_prev  = start_curr - timedelta(days=1)
    start_prev = end_prev - timedelta(days=cfg['days'])

    print(f"GSC Keyword Monitor — {cfg['site_url']}")
    print(f"Periodo corrente:   {start_curr} → {end_curr}")
    print(f"Periodo precedente: {start_prev} → {end_prev}\n")

    # Autenticazione
    service = get_service(cfg['key_file'], cfg['scopes'])

    # Scarica dati
    print("Scaricando dati periodo corrente...")
    rows_curr = fetch_search_analytics(service, cfg['site_url'],
                                       start_curr, end_curr,
                                       row_limit=cfg['row_limit'])
    print(f"  {len(rows_curr)} righe scaricate")

    print("Scaricando dati periodo precedente...")
    rows_prev = fetch_search_analytics(service, cfg['site_url'],
                                       start_prev, end_prev,
                                       row_limit=cfg['row_limit'])
    print(f"  {len(rows_prev)} righe scaricate\n")

    # Converti in DataFrame
    df_curr = rows_to_dataframe(rows_curr)
    df_prev = rows_to_dataframe(rows_prev)

    # Analisi quick wins
    quick_wins = calculate_potential(
        df_curr, cfg['pos_min'], cfg['pos_max'],
        cfg['min_imps'], CTR_BENCHMARK
    )

    # Confronto periodi
    comparison = compare_periods(df_curr, df_prev)

    # Stampa report in console
    print(f"{'='*65}")
    print(f"QUICK WINS (pos {cfg['pos_min']}-{cfg['pos_max']}, min {cfg['min_imps']} imp)")
    print(f"{'='*65}")
    if quick_wins.empty:
        print("  Nessun quick win trovato con i filtri correnti.")
    else:
        print(f"  {'QUERY':<38} {'POS':>5} {'IMP':>6} {'GAIN':>6}  PAGINA")
        print(f"  {'-'*75}")
        for _, row in quick_wins.head(15).iterrows():
            q = row['query'][:36] + '..' if len(row['query']) > 36 else row['query']
            page = row.get('page', '').split('.com')[-1] or '/'
            print(f"  {q:<38} {row['position']:>5.1f} {int(row['impressions']):>6} "
                  f"{row['gain_clic_stimato']:>5}+  {page}")
        print(f"\n  Totale: {len(quick_wins)} opportunità")
        print(f"  Gain totale stimato: +{quick_wins['gain_clic_stimato'].sum()} clic/mese")

    if not comparison.empty:
        rising  = comparison[comparison['delta_pos'] > 1].head(5)
        falling = comparison[comparison['delta_pos'] < -1].head(5)
        if not rising.empty:
            print(f"\n🟢 IN SALITA:")
            for _, row in rising.iterrows():
                print(f"   '{row['query']}' {row['pos_prev']:.1f}→{row['pos_curr']:.1f} ({row['delta_pos']:+.1f})")
        if not falling.empty:
            print(f"\n🔴 IN DISCESA:")
            for _, row in falling.iterrows():
                print(f"   '{row['query']}' {row['pos_prev']:.1f}→{row['pos_curr']:.1f} ({row['delta_pos']:+.1f})")

    # Export
    print(f"\nExport report...")
    export_reports(quick_wins, comparison, cfg['output_dir'], cfg['site_url'])
    print("\nDone.")

if __name__ == '__main__':
    main()

crontab — automazione settimanale

# Ogni lunedì alle 8:00 — genera report e invia notifica Telegram
0 8 * * 1 cd /path/to/scripts && python3 gsc_keyword_monitor.py && \
  python3 -c "
import json, requests
from pathlib import Path
from datetime import date

summary_file = f'gsc_reports/summary_{date.today()}.json'
if Path(summary_file).exists():
    data = json.load(open(summary_file))
    msg = f'📊 GSC Weekly Report\n'
    msg += f'Sito: {data[\"site\"]}\n'
    msg += f'Quick wins: {data[\"quick_wins_count\"]}\n'
    msg += f'Gain stimato: +{data[\"total_gain_potential\"]} clic/mese\n\n'
    for opp in data[\"top_opportunities\"][:3]:
        msg += f'  • {opp[\"query\"]} (pos {opp[\"position\"]:.1f}) → +{opp[\"gain_clic_stimato\"]} clic\n'

    BOT_TOKEN = 'IL_TUO_BOT_TOKEN'
    CHAT_ID   = 'IL_TUO_CHAT_ID'
    requests.post(f'https://api.telegram.org/bot{BOT_TOKEN}/sendMessage',
                  json={'chat_id': CHAT_ID, 'text': msg})
"