How to Automate YouTube Upload Webhooks & Cross-Channel Alerts using the API

1from flask import Flask, request, Response
2import requests, hmac, hashlib, xml.etree.ElementTree as ET
3import os, json, time, logging
4from datetime import datetime, timedelta
5
6app = Flask(__name__)
7logging.basicConfig(level=logging.INFO)
8
9PUSH_SECRET = os.environ.get('PUSH_SECRET', '')
10YT_API_KEY = os.environ['YOUTUBE_API_KEY']
11DISCORD_WEBHOOK = os.environ.get('DISCORD_WEBHOOK_URL')
12CALLBACK_URL = os.environ['PUSH_CALLBACK_URL']
13YT_NS = 'http://www.youtube.com/xml/schemas/2015'
14ATOM_NS = 'http://www.w3.org/2005/Atom'
15
16@app.route('/yt/callback', methods=['GET', 'POST'])
17def youtube_callback():
18    if request.method == 'GET':
19        challenge = request.args.get('hub.challenge')
20        if challenge:
21            return Response(challenge, status=200, mimetype='text/plain')
22        return Response('Bad request', status=400)
23    
24    if PUSH_SECRET:
25        sig = request.headers.get('X-Hub-Signature', '')
26        expected = 'sha1=' + hmac.new(PUSH_SECRET.encode(), request.data, hashlib.sha1).hexdigest()
27        if not hmac.compare_digest(sig, expected):
28            return Response('Forbidden', status=403)
29    
30    try:
31        root = ET.fromstring(request.data.decode())
32        entry = root.find(f'{{{ATOM_NS}}}entry')
33        if not entry:
34            return Response('OK', status=200)
35        video_id = entry.find(f'{{{YT_NS}}}videoId').text
36        channel_id = entry.find(f'{{{YT_NS}}}channelId').text
37        logging.info(f'Upload notification: videoId={video_id} channelId={channel_id}')
38        process_upload_notification(video_id, channel_id)
39    except Exception as e:
40        logging.error(f'Error processing notification: {e}')
41    
42    return Response('OK', status=200)
43
44def process_upload_notification(video_id, channel_id):
45    r = requests.get('https://www.googleapis.com/youtube/v3/videos',
46                     params={'part': 'snippet', 'id': video_id, 'key': YT_API_KEY})
47    items = r.json().get('items', [])
48    if not items:
49        return
50    snippet = items[0]['snippet']
51    title = snippet.get('title', 'Untitled')
52    description = snippet.get('description', '')[:200]
53    thumbnail = snippet.get('thumbnails', {}).get('high', {}).get('url', '')
54    channel_name = snippet.get('channelTitle', channel_id)
55    if DISCORD_WEBHOOK:
56        requests.post(DISCORD_WEBHOOK, json={'embeds': [{
57            'title': title, 'url': f'https://youtu.be/{video_id}',
58            'description': description, 'image': {'url': thumbnail}, 'color': 16711680
59        }]})
60        logging.info(f'Discord alert sent for {title}')
61
62def subscribe(channel_id):
63    requests.post('https://pubsubhubbub.appspot.com/subscribe', data={
64        'hub.callback': CALLBACK_URL, 'hub.mode': 'subscribe',
65        'hub.topic': f'https://www.youtube.com/xml/feeds/videos.xml?channel_id={channel_id}',
66        'hub.verify': 'async', 'hub.lease_seconds': '864000',
67        **({'hub.secret': PUSH_SECRET} if PUSH_SECRET else {})
68    })
69
70if __name__ == '__main__':
71    CHANNELS = os.environ.get('YOUTUBE_CHANNEL_IDS', '').split(',')
72    for ch in CHANNELS:
73        if ch.strip():
74            subscribe(ch.strip())
75    app.run(port=int(os.environ.get('PORT', 3000)))