如何建立历史 SERP 档案

import sqlite3, json, os, requests
from datetime import datetime

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

db = sqlite3.connect('serp_archive.db')
db.execute('''CREATE TABLE IF NOT EXISTS snapshots (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    keyword TEXT NOT NULL,
    captured_at TEXT NOT NULL,
    top_3 TEXT,
    has_ai_overview INTEGER,
    has_featured_snippet INTEGER,
    paa_count INTEGER,
    full_json TEXT
)''')
db.execute('CREATE INDEX IF NOT EXISTS idx_kw_date ON snapshots(keyword, captured_at)')
db.commit()
print('Archive database ready.')

def capture(keyword):
    data = requests.post('https://api.scavio.dev/api/v1/search',
        headers=SH, json={'query': keyword, 'country_code': 'us'}).json()
    organic = data.get('organic_results', [])[:10]
    top_3 = json.dumps([{'pos': r['position'], 'title': r['title'][:60],
        'domain': r['link'].split('/')[2]} for r in organic[:3]])
    has_ao = 1 if data.get('ai_overview') else 0
    has_fs = 1 if data.get('answer_box') else 0
    paa = len(data.get('related_questions', []))
    db.execute('INSERT INTO snapshots (keyword, captured_at, top_3, has_ai_overview, has_featured_snippet, paa_count, full_json) VALUES (?, ?, ?, ?, ?, ?, ?)',
        (keyword, datetime.now().isoformat(), top_3, has_ao, has_fs, paa, json.dumps(data)))
    db.commit()
    print(f'  {keyword}: top={organic[0]["title"][:40] if organic else "N/A"}, AO={has_ao}, FS={has_fs}, PAA={paa}')
    return data

keywords = ['best serp api 2026', 'python web scraping', 'tiktok analytics tool']
for kw in keywords: capture(kw)
print(f'\nCaptured {len(keywords)} snapshots. Cost: ${len(keywords) * 0.005:.3f}')

def timeline(keyword, days=30):
    rows = db.execute(
        'SELECT captured_at, top_3, has_ai_overview, has_featured_snippet, paa_count FROM snapshots WHERE keyword = ? ORDER BY captured_at DESC LIMIT ?',
        (keyword, days)).fetchall()
    if not rows:
        print(f'No data for "{keyword}"')
        return
    print(f'\nTimeline for "{keyword}" ({len(rows)} snapshots):')
    prev_top = None
    for date, top_3, ao, fs, paa in rows:
        top = json.loads(top_3)
        top_domain = top[0]['domain'] if top else 'N/A'
        changed = ' CHANGED' if prev_top and prev_top != top_domain else ''
        print(f'  {date[:10]}: #1={top_domain:30} AO={ao} FS={fs} PAA={paa}{changed}')
        prev_top = top_domain

def feature_report(keyword):
    rows = db.execute(
        'SELECT COUNT(*), SUM(has_ai_overview), SUM(has_featured_snippet), AVG(paa_count) FROM snapshots WHERE keyword = ?',
        (keyword,)).fetchone()
    total, ao_count, fs_count, avg_paa = rows
    print(f'\n"{keyword}": {total} snapshots, AI Overview {ao_count}/{total} ({ao_count/total*100:.0f}%), Featured Snippet {fs_count}/{total} ({fs_count/total*100:.0f}%), Avg PAA: {avg_paa:.1f}')

timeline('best serp api 2026')
feature_report('best serp api 2026')

def daily_capture(keywords_file='keywords.txt'):
    try:
        with open(keywords_file) as f:
            keywords = [line.strip() for line in f if line.strip()]
    except FileNotFoundError:
        keywords = ['best serp api', 'tiktok analytics', 'web scraping tool']
        with open(keywords_file, 'w') as f:
            f.write('\n'.join(keywords))
        print(f'Created {keywords_file} with {len(keywords)} default keywords')

    print(f'Daily capture: {len(keywords)} keywords at {datetime.now().isoformat()}')
    for kw in keywords: capture(kw)
    cost = len(keywords) * 0.005
    print(f'Done. Cost: ${cost:.3f} ({len(keywords)} queries)')
    print(f'Monthly estimate: ${cost * 30:.2f}')
    return len(keywords)

# Run: python archive.py
# Cron: 0 6 * * * cd /path/to/project && python archive.py
daily_capture()