如何构建 AI 模式可见性仪表板

import os, requests, json
from datetime import datetime, timedelta

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

KEYWORDS = [
    'best search api for agents',
    'mcp search tool setup',
    'web search api pricing 2026',
    'serp api for ai apps',
    'search api free tier',
]
BRAND = 'Scavio'
DB_FILE = 'ai_mode_dashboard.json'

def scan_keyword(keyword, brand):
    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 = data.get('organic_results', [])
    brand_l = brand.lower()
    return {
        'keyword': keyword,
        'has_ai': bool(ai),
        'cited_in_ai': brand_l in json.dumps(ai).lower() if ai else False,
        'organic_pos': next((i+1 for i, r in enumerate(organic) if brand_l in r.get('link', '').lower()), None),
        'top_cited': [r.get('displayed_link', '')[:30] for r in organic[:3]],
    }

def daily_scan():
    today = datetime.now().strftime('%Y-%m-%d')
    scans = [scan_keyword(kw, BRAND) for kw in KEYWORDS]
    cited = sum(1 for s in scans if s['cited_in_ai'])
    has_ai = sum(1 for s in scans if s['has_ai'])
    score = (cited / has_ai * 100) if has_ai else 0
    record = {'date': today, 'score': score, 'cited': cited, 'total_ai': has_ai, 'scans': scans}
    # Append to history
    try:
        with open(DB_FILE) as f:
            history = json.load(f)
    except FileNotFoundError:
        history = []
    history.append(record)
    with open(DB_FILE, 'w') as f:
        json.dump(history, f, indent=2)
    print(f'Daily scan: {today} | Score: {score:.0f}% | Cited: {cited}/{has_ai}')
    return record

record = daily_scan()

def analyze_trends(db_file=DB_FILE):
    with open(db_file) as f:
        history = json.load(f)
    if len(history) < 2:
        print('  Need at least 2 days of data for trends.')
        return
    print(f'\n=== AI Mode Visibility Trends ===')
    print(f'  Period: {history[0]["date"]} to {history[-1]["date"]} ({len(history)} days)')
    # Score trend
    scores = [h['score'] for h in history]
    latest = scores[-1]
    prev = scores[-2]
    avg_7d = sum(scores[-7:]) / min(len(scores), 7)
    delta = latest - prev
    direction = 'UP' if delta > 0 else 'DOWN' if delta < 0 else 'STABLE'
    print(f'\n  Today:  {latest:.0f}%')
    print(f'  Change: {delta:+.0f}% ({direction})')
    print(f'  7d avg: {avg_7d:.0f}%')
    # Keyword-level changes
    today_scans = {s['keyword']: s for s in history[-1]['scans']}
    prev_scans = {s['keyword']: s for s in history[-2]['scans']}
    print(f'\n  Keyword Changes:')
    for kw, scan in today_scans.items():
        prev_scan = prev_scans.get(kw, {})
        if scan['cited_in_ai'] != prev_scan.get('cited_in_ai', False):
            change = 'GAINED' if scan['cited_in_ai'] else 'LOST'
            print(f'    {change}: {kw[:40]}')
    # Chart (ASCII)
    print(f'\n  Score History:')
    for h in history[-14:]:
        bar = '#' * int(h['score'] / 5)
        print(f'    {h["date"]} | {bar:20} {h["score"]:.0f}%')

analyze_trends()

from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/api/dashboard')
def dashboard():
    try:
        with open(DB_FILE) as f:
            history = json.load(f)
    except FileNotFoundError:
        return jsonify({'error': 'No data yet. Run daily_scan() first.'})
    latest = history[-1] if history else {}
    scores = [{'date': h['date'], 'score': h['score']} for h in history]
    return jsonify({
        'latest_score': latest.get('score', 0),
        'latest_date': latest.get('date', ''),
        'cited': latest.get('cited', 0),
        'total_ai': latest.get('total_ai', 0),
        'history': scores[-30:],
        'keywords': latest.get('scans', []),
        'cost_per_day': f'${len(KEYWORDS) * 0.005:.3f}',
    })

@app.route('/api/gaps')
def gaps():
    with open(DB_FILE) as f:
        history = json.load(f)
    latest = history[-1] if history else {}
    gap_keywords = [s['keyword'] for s in latest.get('scans', []) if s.get('has_ai') and not s.get('cited_in_ai')]
    return jsonify({'gaps': gap_keywords, 'count': len(gap_keywords)})

# Uncomment to run:
# app.run(port=5050)
print('Dashboard API ready on :5050')
print('  GET /api/dashboard - Overview and trends')
print('  GET /api/gaps - Keywords where AI Mode skips you')
print(f'  Daily cost: ${len(KEYWORDS) * 0.005:.3f}')