Cómo crear un flujo de trabajo de alertas de TikTok n8n

// n8n HTTP Request Node Configuration:
// Method: POST
// URL: https://api.scavio.dev/api/v1/tiktok/profile
// Headers:
//   Authorization: Bearer {{ $env.SCAVIO_API_KEY }}
//   Content-Type: application/json
// Body (JSON):
//   { "username": "{{ $json.username }}" }

// Test with Python to verify response format:
import os, requests
TH = {'Authorization': f'Bearer {os.environ["SCAVIO_API_KEY"]}', 'Content-Type': 'application/json'}
data = requests.post('https://api.scavio.dev/api/v1/tiktok/profile',
    headers=TH, json={'username': 'competitor1'}).json()
user = data.get('user', data.get('data', {}).get('user', data))
print(f'Profile fields available: {list(user.keys())[:8]}')
print(f'Followers: {user.get("followerCount", 0):,}')

// n8n Workflow Structure:
// 1. Schedule Trigger (every 6 hours)
// 2. Set node: Define competitor usernames
// 3. Loop: For each username
//    a. HTTP Request: GET profile
//    b. HTTP Request: GET latest posts
//    c. Google Sheets: Read last known post count
//    d. IF node: New posts detected?
//    e. Google Sheets: Write new metrics
//    f. Gmail: Send alert (if new posts)

// Step 2 - Set node config:
// competitors = ['competitor1', 'competitor2', 'competitor3']

// Step 3a - Profile fetch:
import os, requests
TH = {'Authorization': f'Bearer {os.environ["SCAVIO_API_KEY"]}', 'Content-Type': 'application/json'}

def get_metrics(username):
    p = requests.post('https://api.scavio.dev/api/v1/tiktok/profile',
        headers=TH, json={'username': username}).json()
    user = p.get('user', p.get('data', {}).get('user', p))
    sec_uid = user.get('secUid', '')
    posts = requests.post('https://api.scavio.dev/api/v1/tiktok/user/posts',
        headers=TH, json={'sec_user_id': sec_uid}).json()
    vids = posts.get('videos', posts.get('data', {}).get('videos', []))[:5]
    return {
        'username': username,
        'followers': user.get('followerCount', 0),
        'videos': user.get('videoCount', 0),
        'latest_desc': vids[0].get('desc', '')[:60] if vids else '',
        'latest_plays': vids[0].get('stats', {}).get('playCount', 0) if vids else 0
    }

print(get_metrics('competitor1'))

// Google Sheets structure:
// Sheet: TikTok Monitoring
// Columns: Timestamp | Username | Followers | Video Count | Latest Post | Plays

// n8n Google Sheets Append node:
// Spreadsheet: TikTok Monitoring
// Sheet: Sheet1
// Mapping:
//   Timestamp: {{ $now.toISO() }}
//   Username: {{ $json.username }}
//   Followers: {{ $json.followers }}
//   Video Count: {{ $json.videos }}
//   Latest Post: {{ $json.latest_desc }}
//   Plays: {{ $json.latest_plays }}

# Python equivalent for testing the data flow:
from datetime import datetime

def format_for_sheets(metrics):
    return {
        'timestamp': datetime.now().isoformat(),
        'username': metrics['username'],
        'followers': metrics['followers'],
        'video_count': metrics['videos'],
        'latest_post': metrics['latest_desc'],
        'plays': metrics['latest_plays']
    }

row = format_for_sheets(get_metrics('competitor1'))
print('Sheet row:')
for k, v in row.items():
    print(f'  {k}: {v}')

// n8n Gmail Send node:
// To: your@email.com
// Subject: TikTok Alert: {{ $json.username }} posted new content
// Body:
//   Competitor: @{{ $json.username }}
//   New post: {{ $json.latest_desc }}
//   Plays so far: {{ $json.latest_plays }}
//   Total followers: {{ $json.followers }}
//   Check it: https://tiktok.com/@{{ $json.username }}

# Python test for the alert logic:
def check_and_alert(username, last_known_count):
    metrics = get_metrics(username)
    if metrics['videos'] > last_known_count:
        new_posts = metrics['videos'] - last_known_count
        print(f'ALERT: @{username} posted {new_posts} new video(s)')
        print(f'  Latest: {metrics["latest_desc"]}')
        print(f'  Plays: {metrics["latest_plays"]:,}')
        print(f'  Followers: {metrics["followers"]:,}')
        return True
    else:
        print(f'@{username}: No new posts (still at {last_known_count} videos)')
        return False

# Simulate checking 3 competitors
for comp in ['competitor1', 'competitor2', 'competitor3']:
    check_and_alert(comp, 100)  # Replace 100 with actual last known count
print(f'\nCost per check cycle: ${3 * 0.010:.3f} (profile + posts per competitor)')