如何使用一个 API 构建跨平台品牌监控

import requests, os

API_KEY = os.environ['SCAVIO_API_KEY']
SEARCH_URL = 'https://api.scavio.dev/api/v1/search'
TIKTOK_URL = 'https://api.scavio.dev/api/v1/tiktok'

def search_api(body: dict) -> dict:
    resp = requests.post(SEARCH_URL,
        headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
        json=body)
    resp.raise_for_status()
    return resp.json()

def tiktok_api(endpoint: str, body: dict) -> dict:
    resp = requests.post(f'{TIKTOK_URL}/{endpoint}',
        headers={'Authorization': f'Bearer {API_KEY}',
                 'Content-Type': 'application/json'},
        json=body)
    resp.raise_for_status()
    return resp.json()

def monitor_google(brand: str) -> dict:
    data = search_api({'query': f'"{brand}"', 'country_code': 'us'})
    organic = data.get('organic_results', [])
    return {'platform': 'google', 'mentions': len(organic),
            'top_result': organic[0]['title'] if organic else None}

def monitor_amazon(brand: str) -> dict:
    data = search_api({'platform': 'amazon', 'query': brand, 'marketplace': 'US'})
    products = data.get('products', [])
    return {'platform': 'amazon', 'listings': len(products),
            'avg_rating': sum(p.get('rating', 0) for p in products) / max(len(products), 1)}

def monitor_youtube(brand: str) -> dict:
    data = search_api({'platform': 'youtube', 'query': brand})
    videos = data.get('videos', [])
    return {'platform': 'youtube', 'videos': len(videos),
            'top_video': videos[0].get('title') if videos else None}

for fn in [monitor_google, monitor_amazon, monitor_youtube]:
    result = fn('yourBrand')
    print(result)

def monitor_tiktok(brand: str) -> dict:
    # Video mentions
    search_data = tiktok_api('search/videos',
        {'keyword': brand, 'count': 20, 'cursor': 0})['data']
    video_count = len(search_data.get('videos', []))
    total_plays = sum(v.get('stats', {}).get('playCount', 0)
                      for v in search_data.get('videos', []))
    # Hashtag presence
    try:
        hashtag_data = tiktok_api('hashtag', {'hashtag': brand})['data']
        hashtag_views = hashtag_data['stats']['view_count']
    except Exception:
        hashtag_views = 0
    return {
        'platform': 'tiktok',
        'video_mentions': video_count,
        'total_plays': total_plays,
        'hashtag_views': hashtag_views
    }

tiktok_stats = monitor_tiktok('yourBrand')
print(tiktok_stats)

import json
from datetime import date

def brand_report(brands: list) -> dict:
    report = {'date': date.today().isoformat(), 'brands': {}}
    for brand in brands:
        report['brands'][brand] = {
            'google': monitor_google(brand),
            'amazon': monitor_amazon(brand),
            'youtube': monitor_youtube(brand),
            'tiktok': monitor_tiktok(brand)
        }
    # Share of voice: total mentions across platforms
    for brand in brands:
        data = report['brands'][brand]
        total = (data['google']['mentions'] + data['amazon']['listings'] +
                 data['youtube']['videos'] + data['tiktok']['video_mentions'])
        data['total_mentions'] = total
    grand_total = sum(b['total_mentions'] for b in report['brands'].values())
    for brand in brands:
        sov = report['brands'][brand]['total_mentions'] / max(grand_total, 1)
        report['brands'][brand]['share_of_voice'] = round(sov * 100, 1)
    with open(f'brand_monitor_{date.today()}.json', 'w') as f:
        json.dump(report, f, indent=2)
    return report

report = brand_report(['yourBrand', 'competitorA', 'competitorB'])
for brand, data in report['brands'].items():
    print(f"{brand}: {data['total_mentions']} mentions, {data['share_of_voice']}% SoV")

def estimate_monthly_cost(num_brands: int, runs_per_day: int = 1) -> dict:
    # Per brand per run: 1 Google + 1 Amazon + 1 YouTube + 2 TikTok = 5 credits
    credits_per_run = num_brands * 5
    daily_credits = credits_per_run * runs_per_day
    monthly_credits = daily_credits * 30
    cost = monthly_credits * 0.005
    return {
        'brands': num_brands,
        'daily_credits': daily_credits,
        'monthly_credits': monthly_credits,
        'monthly_cost': f'${cost:.2f}',
        'recommended_plan': '$30/7K credits' if monthly_credits <= 7000
            else '$100/28K credits' if monthly_credits <= 28000
            else '$250/85K credits'
    }

for n in [3, 10, 25]:
    est = estimate_monthly_cost(n)
    print(f"{est['brands']} brands: {est['monthly_credits']} credits/mo "
          f"({est['monthly_cost']}), plan: {est['recommended_plan']}")