如何检测 Google I/O 之后 AI 概述的变化

import os, requests, json, hashlib
from datetime import datetime

API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}

KEYWORDS = ['best search api 2026', 'how to add search to ai agent', 'mcp tools for agents']

def snapshot_ai_overview(keyword):
    data = requests.post('https://api.scavio.dev/api/v1/search',
        headers=SH, json={'query': keyword, 'country_code': 'us'}, timeout=10).json()
    ai = data.get('ai_overview', data.get('answer_box', {}))
    organic_top3 = [r.get('link', '') for r in data.get('organic_results', [])[:3]]
    featured = data.get('featured_snippet', {})
    content_str = json.dumps(ai, sort_keys=True)
    return {
        'keyword': keyword,
        'timestamp': datetime.now().isoformat(),
        'has_ai_overview': bool(ai),
        'ai_content': ai,
        'ai_content_hash': hashlib.md5(content_str.encode()).hexdigest(),
        'organic_top3': organic_top3,
        'has_featured': bool(featured),
    }

today = []
for kw in KEYWORDS:
    snap = snapshot_ai_overview(kw)
    today.append(snap)
    print(f'  {kw[:40]:40} | AI: {"yes" if snap["has_ai_overview"] else "no":3} | hash: {snap["ai_content_hash"][:8]}')
print(f'\nSnapshots: {len(today)} | Cost: ${len(KEYWORDS) * 0.005:.3f}')

HISTORY_FILE = 'ai_overview_history.json'

def load_history():
    try:
        with open(HISTORY_FILE) as f:
            return json.load(f)
    except FileNotFoundError:
        return []

def save_history(history):
    with open(HISTORY_FILE, 'w') as f:
        json.dump(history, f, indent=2)

def detect_changes(today_snaps, history):
    if not history:
        print('  First snapshot. No comparison available.')
        return []
    prev_day = history[-1]
    prev_by_kw = {s['keyword']: s for s in prev_day['snapshots']}
    changes = []
    for snap in today_snaps:
        kw = snap['keyword']
        prev = prev_by_kw.get(kw)
        if not prev:
            continue
        change = {'keyword': kw, 'changes': []}
        if snap['ai_content_hash'] != prev['ai_content_hash']:
            change['changes'].append('AI Overview content changed')
        if snap['has_ai_overview'] != prev['has_ai_overview']:
            status = 'appeared' if snap['has_ai_overview'] else 'disappeared'
            change['changes'].append(f'AI Overview {status}')
        if snap['organic_top3'] != prev['organic_top3']:
            change['changes'].append('Top 3 organic results changed')
        if change['changes']:
            changes.append(change)
            for c in change['changes']:
                print(f'  CHANGE: {kw[:35]} -> {c}')
    if not changes:
        print('  No changes detected.')
    return changes

history = load_history()
changes = detect_changes(today, history)
history.append({'date': datetime.now().strftime('%Y-%m-%d'), 'snapshots': today})
save_history(history)

def change_report(changes, today_snaps):
    print(f'\n{"=" * 60}')
    print(f'  AI Overview Change Report - {datetime.now().strftime("%Y-%m-%d")}')
    print(f'  Post Google I/O 2026 Monitoring')
    print(f'{"=" * 60}')
    print(f'\n  Keywords monitored: {len(today_snaps)}')
    print(f'  Changes detected: {len(changes)}')
    ai_count = sum(1 for s in today_snaps if s['has_ai_overview'])
    print(f'  AI Overviews present: {ai_count}/{len(today_snaps)}')
    if changes:
        print(f'\n  Changes:')
        for c in changes:
            print(f'    {c["keyword"][:40]}')
            for ch in c['changes']:
                print(f'      - {ch}')
    # Alert levels
    ai_changes = [c for c in changes if any('AI Overview' in ch for ch in c['changes'])]
    if ai_changes:
        print(f'\n  ALERT: {len(ai_changes)} AI Overview structure changes detected.')
        print(f'  This may indicate post-I/O algorithm updates.')
        print(f'  Review affected keywords and update content strategy.')
    print(f'\n  Daily cost: ${len(today_snaps) * 0.005:.3f}')
    print(f'  Monthly: ${len(today_snaps) * 0.005 * 30:.2f}')

change_report(changes, today)