diff --git a/database/models.py b/database/models.py
index 9f103d1..e174fce 100644
--- a/database/models.py
+++ b/database/models.py
@@ -61,3 +61,13 @@ class AntiCheatCache(db.Model):
notes = db.Column(db.String(1024))
updated_at = db.Column(db.DateTime)
+class YouTubeNotification(db.Model):
+ __tablename__ = 'youtube_notification'
+ id = db.Column(db.Integer, primary_key=True)
+ enable = db.Column(db.Boolean, default=True)
+ channel_id = db.Column(db.String(128)) # ID de la chaîne YouTube
+ notify_channel = db.Column(db.Integer) # ID du canal Discord
+ message = db.Column(db.String(2000))
+ video_type = db.Column(db.String(16), default='all') # 'all', 'video', 'short'
+ last_video_id = db.Column(db.String(128)) # ID de la dernière vidéo notifiée
+
diff --git a/database/schema.sql b/database/schema.sql
index 04cef86..4f36e26 100644
--- a/database/schema.sql
+++ b/database/schema.sql
@@ -76,3 +76,13 @@ CREATE TABLE IF NOT EXISTS `member_invites` (
`inviter_name` VARCHAR(256),
`join_date` DATETIME NOT NULL
);
+
+CREATE TABLE IF NOT EXISTS `youtube_notification` (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ `enable` BOOLEAN NOT NULL DEFAULT TRUE,
+ `channel_id` VARCHAR(128) NOT NULL,
+ `notify_channel` INTEGER NOT NULL,
+ `message` VARCHAR(2000) NOT NULL,
+ `video_type` VARCHAR(16) NOT NULL DEFAULT 'all',
+ `last_video_id` VARCHAR(128)
+);
diff --git a/discordbot/__init__.py b/discordbot/__init__.py
index 2de81dd..9c48341 100644
--- a/discordbot/__init__.py
+++ b/discordbot/__init__.py
@@ -22,6 +22,7 @@ from discordbot.moderation import (
handle_say_command
)
from discordbot.welcome import sendWelcomeMessage, sendLeaveMessage, updateInviteCache
+from discordbot.youtube import checkYouTubeVideos
from protondb import searhProtonDb
class DiscordBot(discord.Client):
@@ -35,6 +36,7 @@ class DiscordBot(discord.Client):
self.loop.create_task(self.updateStatus())
self.loop.create_task(self.updateHumbleBundle())
+ self.loop.create_task(self.updateYouTube())
async def updateStatus(self):
while not self.is_closed():
@@ -50,6 +52,12 @@ class DiscordBot(discord.Client):
while not self.is_closed():
await checkHumbleBundleAndNotify(self)
await asyncio.sleep(30*60)
+
+ async def updateYouTube(self):
+ while not self.is_closed():
+ await checkYouTubeVideos()
+ # Vérification toutes les 5 minutes (comme pour Twitch)
+ await asyncio.sleep(5*60)
def getAllTextChannel(self) -> list[TextChannel]:
channels = []
diff --git a/discordbot/youtube.py b/discordbot/youtube.py
new file mode 100644
index 0000000..7211fca
--- /dev/null
+++ b/discordbot/youtube.py
@@ -0,0 +1,180 @@
+import logging
+import asyncio
+import xml.etree.ElementTree as ET
+import requests
+
+from database import db
+from database.models import YouTubeNotification
+from webapp import webapp
+
+logger = logging.getLogger('youtube-notification')
+logger.setLevel(logging.INFO)
+
+
+async def checkYouTubeVideos():
+ with webapp.app_context():
+ try:
+ notifications: list[YouTubeNotification] = YouTubeNotification.query.filter_by(enable=True).all()
+
+ for notification in notifications:
+ try:
+ await _checkChannelVideos(notification)
+ except Exception as e:
+ logger.error(f"Erreur lors de la vérification de la chaîne {notification.channel_id}: {e}")
+ continue
+ except Exception as e:
+ logger.error(f"Erreur lors de la vérification YouTube: {e}")
+
+
+async def _checkChannelVideos(notification: YouTubeNotification):
+ try:
+ channel_id = notification.channel_id
+
+ rss_url = f"https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"
+
+ response = await asyncio.to_thread(requests.get, rss_url, timeout=10)
+
+ if response.status_code != 200:
+ logger.error(f"Erreur HTTP {response.status_code} lors de la récupération du RSS pour {channel_id}")
+ return
+
+ root = ET.fromstring(response.content)
+
+ ns = {'atom': 'http://www.w3.org/2005/Atom', 'yt': 'http://www.youtube.com/xml/schemas/2015', 'media': 'http://search.yahoo.com/mrss/'}
+
+ entries = root.findall('atom:entry', ns)
+
+ if not entries:
+ logger.warning(f"Aucune vidéo trouvée dans le RSS pour {channel_id}")
+ return
+
+ videos = []
+ for entry in entries:
+ video_id = entry.find('yt:videoId', ns)
+ if video_id is None:
+ continue
+ video_id = video_id.text
+
+ title_elem = entry.find('atom:title', ns)
+ video_title = title_elem.text if title_elem is not None else 'Sans titre'
+
+ link_elem = entry.find('atom:link', ns)
+ video_url = link_elem.get('href') if link_elem is not None else f"https://www.youtube.com/watch?v={video_id}"
+
+ published_elem = entry.find('atom:published', ns)
+ published_at = published_elem.text if published_elem is not None else ''
+
+ author_elem = entry.find('atom:author/atom:name', ns)
+ channel_name = author_elem.text if author_elem is not None else 'Inconnu'
+
+ thumbnail = None
+ media_thumbnail = entry.find('media:group/media:thumbnail', ns)
+ if media_thumbnail is not None:
+ thumbnail = media_thumbnail.get('url')
+
+ is_short = False
+ if video_title and ('#shorts' in video_title.lower() or '#short' in video_title.lower()):
+ is_short = True
+
+ if notification.video_type == 'all':
+ videos.append((video_id, {
+ 'title': video_title,
+ 'url': video_url,
+ 'published': published_at,
+ 'channel_name': channel_name,
+ 'thumbnail': thumbnail,
+ 'is_short': is_short
+ }))
+ elif notification.video_type == 'short' and is_short:
+ videos.append((video_id, {
+ 'title': video_title,
+ 'url': video_url,
+ 'published': published_at,
+ 'channel_name': channel_name,
+ 'thumbnail': thumbnail,
+ 'is_short': is_short
+ }))
+ elif notification.video_type == 'video' and not is_short:
+ videos.append((video_id, {
+ 'title': video_title,
+ 'url': video_url,
+ 'published': published_at,
+ 'channel_name': channel_name,
+ 'thumbnail': thumbnail,
+ 'is_short': is_short
+ }))
+
+ videos.sort(key=lambda x: x[1]['published'], reverse=True)
+
+ if videos:
+ latest_video_id, latest_video = videos[0]
+
+ if not notification.last_video_id:
+ notification.last_video_id = latest_video_id
+ db.session.commit()
+ return
+
+ if latest_video_id != notification.last_video_id:
+ logger.info(f"Nouvelle vidéo détectée: {latest_video_id} pour la chaîne {notification.channel_id}")
+ await _notifyVideo(notification, latest_video, latest_video_id)
+ notification.last_video_id = latest_video_id
+ db.session.commit()
+
+ except Exception as e:
+ logger.error(f"Erreur lors de la vérification des vidéos: {e}")
+
+
+async def _notifyVideo(notification: YouTubeNotification, video_data: dict, video_id: str):
+ from discordbot import bot
+ try:
+ channel_name = video_data.get('channel_name', 'Inconnu')
+ video_title = video_data.get('title', 'Sans titre')
+ video_url = video_data.get('url', f"https://www.youtube.com/watch?v={video_id}")
+ thumbnail = video_data.get('thumbnail', '')
+ published_at = video_data.get('published', '')
+ is_short = video_data.get('is_short', False)
+
+ try:
+ message = notification.message.format(
+ channel_name=channel_name or 'Inconnu',
+ video_title=video_title or 'Sans titre',
+ video_url=video_url,
+ video_id=video_id,
+ thumbnail=thumbnail or '',
+ published_at=published_at or '',
+ is_short=is_short
+ )
+ except KeyError as e:
+ logger.error(f"Variable manquante dans le message de notification: {e}")
+ message = f"🎥 Nouvelle vidéo de {channel_name}: [{video_title}]({video_url})"
+
+ logger.info(f"Envoi de notification YouTube: {message}")
+ bot.loop.create_task(_sendMessage(notification.notify_channel, message, video_url, thumbnail, video_title, channel_name))
+
+ except Exception as e:
+ logger.error(f"Erreur lors de la notification: {e}")
+
+
+async def _sendMessage(channel_id: int, message: str, video_url: str, thumbnail: str, video_title: str, channel_name: str):
+ from discordbot import bot
+ try:
+ discord_channel = bot.get_channel(channel_id)
+ if not discord_channel:
+ logger.error(f"Canal Discord {channel_id} introuvable")
+ return
+
+ import discord
+ embed = discord.Embed(
+ title=video_title,
+ url=video_url,
+ color=0xFF0000
+ )
+ embed.set_author(name=channel_name, icon_url="https://www.youtube.com/img/desktop/yt_1200.png")
+ if thumbnail:
+ embed.set_image(url=thumbnail)
+
+ await discord_channel.send(message, embed=embed)
+ logger.info(f"Notification YouTube envoyée avec succès")
+
+ except Exception as e:
+ logger.error(f"Erreur lors de l'envoi du message Discord: {e}")
diff --git a/webapp/__init__.py b/webapp/__init__.py
index 28f6c3c..cdaaa26 100644
--- a/webapp/__init__.py
+++ b/webapp/__init__.py
@@ -2,4 +2,4 @@ from flask import Flask
webapp = Flask(__name__)
-from webapp import commandes, configurations, index, humeurs, protondb, live_alert, twitch_auth, moderation
+from webapp import commandes, configurations, index, humeurs, protondb, live_alert, twitch_auth, moderation, youtube
diff --git a/webapp/templates/template.html b/webapp/templates/template.html
index e36f520..1539260 100644
--- a/webapp/templates/template.html
+++ b/webapp/templates/template.html
@@ -19,6 +19,7 @@
Alerte live YouTube Commandes Humeurs Modération Notifications YouTube
+
+{% if msg %}
+
+ {{ msg }}
+
+
+{% endif %}
+
+
+ Liste des chaînes YouTube surveillées pour les notifications de nouvelles vidéos.
+
+ Le bot vérifie toutes les 5 minutes les nouvelles vidéos des chaînes en dessous.
+ Quand une nouvelle vidéo est détectée, le bot enverra une notification sur Discord.
+
+
+{% if not notification %}
+Notifications
+
+
+
+ Chaîne YouTube
+ Canal Discord
+ Type
+ Message
+ #
+
+
+
+ {% for notification in notifications %}
+
+ {{notification.channel_id}}
+ {{notification.notify_channel_name}}
+
+ {% if notification.video_type == 'all' %}
+ Toutes
+ {% elif notification.video_type == 'video' %}
+ Vidéos uniquement
+ {% elif notification.video_type == 'short' %}
+ Shorts uniquement
+ {% endif %}
+
+ {{notification.message}}
+
+ {{ '✅' if notification.enable else '❌' }}
+ ✐
+ 🗑
+
+
+ {% endfor %}
+
+
+{% endif %}
+
+{{ 'Editer une notification' if notification else 'Ajouter une notification YouTube' }}
+
+
+{% endblock %}
diff --git a/webapp/youtube.py b/webapp/youtube.py
new file mode 100644
index 0000000..0e70118
--- /dev/null
+++ b/webapp/youtube.py
@@ -0,0 +1,157 @@
+import re
+import requests
+from urllib.parse import urlencode
+from flask import render_template, request, redirect, url_for
+
+from webapp import webapp
+from database import db
+from database.models import YouTubeNotification
+from discordbot import bot
+
+
+def extract_channel_id(channel_input: str) -> str:
+ """Extrait l'ID de la chaîne YouTube depuis différents formats"""
+ if not channel_input:
+ return None
+
+ channel_input = channel_input.strip()
+
+ if channel_input.startswith('UC') and len(channel_input) == 24:
+ return channel_input
+
+ if '/channel/' in channel_input:
+ match = re.search(r'/channel/([a-zA-Z0-9_-]{24})', channel_input)
+ if match:
+ return match.group(1)
+
+ if '/c/' in channel_input or '/user/' in channel_input:
+ parts = channel_input.split('/')
+ for i, part in enumerate(parts):
+ if part in ['c', 'user'] and i + 1 < len(parts):
+ handle = parts[i + 1].split('?')[0].split('&')[0]
+ channel_id = _get_channel_id_from_handle(handle)
+ if channel_id:
+ return channel_id
+
+ if '@' in channel_input:
+ handle = re.search(r'@([a-zA-Z0-9_-]+)', channel_input)
+ if handle:
+ channel_id = _get_channel_id_from_handle(handle.group(1))
+ if channel_id:
+ return channel_id
+
+ return None
+
+
+def _get_channel_id_from_handle(handle: str) -> str:
+ """Récupère l'ID de la chaîne depuis un handle en utilisant le flux RSS"""
+ try:
+ url = f"https://www.youtube.com/@{handle}"
+ response = requests.get(url, timeout=10, allow_redirects=True)
+
+ if response.status_code == 200:
+ channel_id_match = re.search(r'"channelId":"([^"]{24})"', response.text)
+ if channel_id_match:
+ return channel_id_match.group(1)
+
+ canonical_match = re.search(r'")
+def openEditYouTube(id):
+ notification = YouTubeNotification.query.get_or_404(id)
+ channels = bot.getAllTextChannel()
+ msg = request.args.get('msg')
+ msg_type = request.args.get('type', 'info')
+ return render_template("youtube.html", notification=notification, channels=channels, notifications=YouTubeNotification.query.all(), msg=msg, msg_type=msg_type)
+
+
+@webapp.route("/youtube/edit/", methods=['POST'])
+def submitEditYouTube(id):
+ notification: YouTubeNotification = YouTubeNotification.query.get_or_404(id)
+
+ channel_input = request.form.get('channel_id', '').strip()
+ channel_id = extract_channel_id(channel_input)
+
+ if not channel_id:
+ return redirect(url_for("openEditYouTube", id=id) + "?" + urlencode({'msg': f"Impossible d'extraire l'ID de la chaîne depuis : {channel_input}. Veuillez vérifier le lien.", 'type': 'error'}))
+
+ notify_channel_str = request.form.get('notify_channel')
+ if not notify_channel_str:
+ return redirect(url_for("openEditYouTube", id=id) + "?" + urlencode({'msg': "Veuillez sélectionner un canal Discord. Assurez-vous que le bot Discord est connecté.", 'type': 'error'}))
+
+ try:
+ notify_channel = int(notify_channel_str)
+ except ValueError:
+ return redirect(url_for("openEditYouTube", id=id) + "?" + urlencode({'msg': "Canal Discord invalide.", 'type': 'error'}))
+
+ notification.channel_id = channel_id
+ notification.notify_channel = notify_channel
+ notification.message = request.form.get('message')
+ notification.video_type = request.form.get('video_type', 'all')
+ db.session.commit()
+ return redirect(url_for("openYouTube") + "?" + urlencode({'msg': "Notification modifiée avec succès", 'type': 'success'}))
+
+
+@webapp.route("/youtube/del/")
+def delYouTube(id):
+ notification = YouTubeNotification.query.get_or_404(id)
+ db.session.delete(notification)
+ db.session.commit()
+ return redirect(url_for("openYouTube"))