3 Commits

9 changed files with 724 additions and 1 deletions

View File

@@ -33,7 +33,13 @@ def _set_sqlite_pragma(dbapi_connection, connection_record):
except Exception:
pass
def _tableExists(table_name:str, cursor:Cursor) -> bool:
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,))
return cursor.fetchone() is not None
def _tableHaveColumn(table_name:str, column_name:str, cursor:Cursor) -> bool:
if not _tableExists(table_name, cursor):
return False
cursor.execute(f'PRAGMA table_info({table_name})')
columns = cursor.fetchall()
return any(col[1] == column_name for col in columns)
@@ -65,6 +71,27 @@ def _doPostImportMigration(cursor:Cursor):
logging.info("suppression de la table temporaire game_bundle_old")
_dropTable('game_bundle_old', cursor)
if _tableExists('youtube_notification', cursor):
logging.info("Migration de la table youtube_notification: ajout des colonnes d'embed")
embed_columns = [
('embed_title', 'VARCHAR(256)'),
('embed_description', 'VARCHAR(2000)'),
('embed_color', 'VARCHAR(8) DEFAULT "FF0000"'),
('embed_footer', 'VARCHAR(2048)'),
('embed_author_name', 'VARCHAR(256)'),
('embed_author_icon', 'VARCHAR(512)'),
('embed_thumbnail', 'BOOLEAN DEFAULT 1'),
('embed_image', 'BOOLEAN DEFAULT 1')
]
for col_name, col_type in embed_columns:
if not _tableHaveColumn('youtube_notification', col_name, cursor):
try:
cursor.execute(f'ALTER TABLE youtube_notification ADD COLUMN {col_name} {col_type}')
logging.info(f"Colonne {col_name} ajoutée à youtube_notification")
except Exception as e:
logging.error(f"Impossible d'ajouter la colonne {col_name}: {e}")
raise
with webapp.app_context():
with open('database/schema.sql', 'r') as f:
sql = f.read()

View File

@@ -61,3 +61,21 @@ 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))
notify_channel = db.Column(db.Integer)
message = db.Column(db.String(2000))
video_type = db.Column(db.String(16), default='all')
last_video_id = db.Column(db.String(128))
embed_title = db.Column(db.String(256))
embed_description = db.Column(db.String(2000))
embed_color = db.Column(db.String(8), default='FF0000')
embed_footer = db.Column(db.String(2048))
embed_author_name = db.Column(db.String(256))
embed_author_icon = db.Column(db.String(512))
embed_thumbnail = db.Column(db.Boolean, default=True)
embed_image = db.Column(db.Boolean, default=True)

View File

@@ -76,3 +76,21 @@ 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),
`embed_title` VARCHAR(256),
`embed_description` VARCHAR(2000),
`embed_color` VARCHAR(8) NOT NULL DEFAULT 'FF0000',
`embed_footer` VARCHAR(2048),
`embed_author_name` VARCHAR(256),
`embed_author_icon` VARCHAR(512),
`embed_thumbnail` BOOLEAN NOT NULL DEFAULT TRUE,
`embed_image` BOOLEAN NOT NULL DEFAULT TRUE
);

View File

@@ -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():
@@ -51,6 +53,12 @@ class DiscordBot(discord.Client):
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 = []
for channel in self.get_all_channels():

225
discordbot/youtube.py Normal file
View File

@@ -0,0 +1,225 @@
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, message, video_url, thumbnail, video_title, channel_name, video_id, published_at, is_short))
except Exception as e:
logger.error(f"Erreur lors de la notification: {e}")
def _format_embed_text(text: str, channel_name: str, video_title: str, video_url: str, video_id: str, thumbnail: str, published_at: str, is_short: bool) -> str:
"""Formate un texte d'embed avec les variables disponibles"""
if not text:
return None
try:
return text.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:
return text
async def _sendMessage(notification: YouTubeNotification, message: str, video_url: str, thumbnail: str, video_title: str, channel_name: str, video_id: str, published_at: str, is_short: bool):
from discordbot import bot
try:
discord_channel = bot.get_channel(notification.notify_channel)
if not discord_channel:
logger.error(f"Canal Discord {notification.notify_channel} introuvable")
return
import discord
embed_title = _format_embed_text(notification.embed_title, channel_name, video_title, video_url, video_id, thumbnail, published_at, is_short) if notification.embed_title else video_title
embed_description = _format_embed_text(notification.embed_description, channel_name, video_title, video_url, video_id, thumbnail, published_at, is_short) if notification.embed_description else None
try:
embed_color = int(notification.embed_color or 'FF0000', 16)
except ValueError:
embed_color = 0xFF0000
embed = discord.Embed(
title=embed_title,
url=video_url,
color=embed_color
)
if embed_description:
embed.description = embed_description
author_name = _format_embed_text(notification.embed_author_name, channel_name, video_title, video_url, video_id, thumbnail, published_at, is_short) if notification.embed_author_name else channel_name
author_icon = notification.embed_author_icon if notification.embed_author_icon else "https://www.youtube.com/img/desktop/yt_1200.png"
embed.set_author(name=author_name, icon_url=author_icon)
if notification.embed_thumbnail and thumbnail:
embed.set_thumbnail(url=thumbnail)
if notification.embed_image and thumbnail:
embed.set_image(url=thumbnail)
if notification.embed_footer:
footer_text = _format_embed_text(notification.embed_footer, channel_name, video_title, video_url, video_id, thumbnail, published_at, is_short)
if footer_text:
embed.set_footer(text=footer_text)
if message and message.strip():
await discord_channel.send(message, embed=embed)
else:
await discord_channel.send(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}")

View File

@@ -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

View File

@@ -19,6 +19,7 @@
<a href="/"><img src="/static/ico/favicon.ico"></a>
<ul>
<li><a href="/live-alert">Alerte live</a></li>
<li><a href="/youtube">YouTube</a></li>
<li><a href="/commandes">Commandes</a></li>
<li><a href="/humeurs">Humeurs</a></li>
<li><a href="/moderation">Modération</a></li>

View File

@@ -0,0 +1,245 @@
{% extends "template.html" %}
{% block content %}
<h1>Notifications YouTube</h1>
{% if msg %}
<div id="alert-msg" class="alert alert-{{ msg_type }}" style="padding: 10px; margin: 10px 0; border: 1px solid {{ '#f00' if msg_type == 'error' else '#0f0' }}; background-color: {{ '#ffe0e0' if msg_type == 'error' else '#e0ffe0' }};">
{{ msg }}
</div>
<script>
setTimeout(function() {
var el = document.getElementById('alert-msg');
if (el) el.style.display = 'none';
}, 5000);
</script>
{% endif %}
<p>
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.
</p>
{% if not notification %}
<h2>Notifications</h2>
<table class="live-alert">
<thead>
<tr>
<th>Chaîne YouTube</th>
<th>Canal Discord</th>
<th>Type</th>
<th>Message</th>
<th>#</th>
</tr>
</thead>
<tbody>
{% for notification in notifications %}
<tr>
<td>{{notification.channel_id}}</td>
<td>{{notification.notify_channel_name}}</td>
<td>
{% if notification.video_type == 'all' %}
Toutes
{% elif notification.video_type == 'video' %}
Vidéos uniquement
{% elif notification.video_type == 'short' %}
Shorts uniquement
{% endif %}
</td>
<td>{{notification.message}}</td>
<td>
<a href="{{ url_for('toggleYouTube', id = notification.id) }}" class="icon">{{ '✅' if notification.enable else '❌' }}</a>
<a href="{{ url_for('openEditYouTube', id = notification.id) }}" class="icon"></a>
<a href="{{ url_for('delYouTube', id = notification.id) }}"
onclick="return confirm('Êtes-vous sûr de vouloir supprimer cette notification ?')" class="icon">🗑</a>
</td>
</tr>
{% endfor %}
</tbody>
</table>
{% endif %}
<h2>{{ 'Editer une notification' if notification else 'Ajouter une notification YouTube' }}</h2>
<div style="display: grid; grid-template-columns: 1fr 1fr; gap: 20px;">
<div>
<form id="youtube-form" action="{{ url_for('submitEditYouTube', id = notification.id) if notification else url_for('addYouTube') }}" method="POST">
<fieldset>
<legend>Configuration de base</legend>
<label for="channel_id">Lien ou ID de la chaîne YouTube</label>
<input name="channel_id" id="channel_id" type="text" maxlength="256" required="required" value="{{notification.channel_id if notification}}" placeholder="https://www.youtube.com/@513v3 ou https://www.youtube.com/channel/UC... ou UC..."/>
<label for="notify_channel">Canal de Notification Discord</label>
<select name="notify_channel" id="notify_channel">
{% for channel in channels %}
<option value="{{channel.id}}"{% if notification and notification.notify_channel == channel.id %}
selected="selected" {% endif %}>{{channel.name}}</option>
{% endfor %}
</select>
<label for="video_type">Type de vidéo à notifier</label>
<select name="video_type" id="video_type">
<option value="all"{% if notification and notification.video_type == 'all' %} selected="selected" {% endif %}>Toutes (vidéos + shorts)</option>
<option value="video"{% if notification and notification.video_type == 'video' %} selected="selected" {% endif %}>Vidéos uniquement</option>
<option value="short"{% if notification and notification.video_type == 'short' %} selected="selected" {% endif %}>Shorts uniquement</option>
</select>
<label for="message">Message (optionnel, envoyé avant l'embed)</label>
<textarea name="message" id="message" rows="3" cols="50">{{notification.message if notification}}</textarea>
</fieldset>
<fieldset>
<legend>Personnalisation de l'embed Discord</legend>
<label for="embed_title">Titre de l'embed</label>
<input name="embed_title" id="embed_title" type="text" maxlength="256" value="{{notification.embed_title if notification}}" placeholder="{video_title} (par défaut: titre de la vidéo)"/>
<small>Variables: {video_title}, {channel_name}, {video_url}, {video_id}</small>
<label for="embed_description">Description de l'embed</label>
<textarea name="embed_description" id="embed_description" rows="4" cols="50" placeholder="Description optionnelle de l'embed">{{notification.embed_description if notification}}</textarea>
<small>Variables: {video_title}, {channel_name}, {video_url}, {published_at}, {is_short}</small>
<label for="embed_color">Couleur de l'embed (hexadécimal)</label>
<input name="embed_color" id="embed_color" type="color" value="#{{notification.embed_color if notification else 'FF0000'}}" style="width: 100px; height: 40px;"/>
<input type="text" id="embed_color_text" value="{{notification.embed_color if notification else 'FF0000'}}" placeholder="FF0000" style="width: 100px; margin-left: 10px;" maxlength="6"/>
<small>Format: FF0000 (rouge YouTube par défaut)</small>
<label for="embed_author_name">Nom de l'auteur</label>
<input name="embed_author_name" id="embed_author_name" type="text" maxlength="256" value="{{notification.embed_author_name if notification}}" placeholder="{channel_name} (par défaut: nom de la chaîne)"/>
<label for="embed_author_icon">Icône de l'auteur (URL)</label>
<input name="embed_author_icon" id="embed_author_icon" type="text" maxlength="512" value="{{notification.embed_author_icon if notification}}" placeholder="https://www.youtube.com/img/desktop/yt_1200.png"/>
<label for="embed_footer">Pied de page</label>
<input name="embed_footer" id="embed_footer" type="text" maxlength="2048" value="{{notification.embed_footer if notification}}" placeholder="Texte optionnel en bas de l'embed"/>
<label>
<input type="checkbox" name="embed_thumbnail" id="embed_thumbnail" {% if not notification or notification.embed_thumbnail %}checked="checked"{% endif %}>
Afficher la miniature (thumbnail) en haut à droite
</label>
<label>
<input type="checkbox" name="embed_image" id="embed_image" {% if not notification or notification.embed_image %}checked="checked"{% endif %}>
Afficher l'image principale (image de la vidéo)
</label>
</fieldset>
<input type="Submit" value="{{ 'Modifier' if notification else 'Ajouter' }}">
</form>
</div>
<div>
<h3>Prévisualisation de l'embed Discord</h3>
<div id="embed-preview" style="background-color: #2f3136; border-radius: 4px; padding: 16px; font-family: 'Whitney', 'Helvetica Neue', Helvetica, Arial, sans-serif; color: #dcddde; max-width: 520px; border-left: 4px solid #FF0000;">
<div id="embed-author" style="display: flex; align-items: center; margin-bottom: 8px; font-size: 14px;">
<img id="embed-author-icon" src="https://www.youtube.com/img/desktop/yt_1200.png" style="width: 20px; height: 20px; border-radius: 50%; margin-right: 8px;" onerror="this.style.display='none'"/>
<span id="embed-author-name" style="font-weight: 600;">Nom de la chaîne</span>
</div>
<a id="embed-title" href="#" style="color: #00aff4; text-decoration: none; font-size: 16px; font-weight: 600; display: block; margin-bottom: 8px;">Titre de la vidéo</a>
<div id="embed-description" style="font-size: 14px; line-height: 1.375; margin-bottom: 8px; color: #dcddde;"></div>
<div id="embed-thumbnail-container" style="margin: 8px 0;">
<img id="embed-thumbnail" src="" style="max-width: 80px; max-height: 80px; border-radius: 4px; float: right; margin-left: 16px; display: none;"/>
</div>
<div id="embed-image-container" style="margin-top: 16px;">
<img id="embed-image" src="https://i.ytimg.com/vi/dQw4w9WgXcQ/maxresdefault.jpg" style="max-width: 100%; border-radius: 4px; display: none;"/>
</div>
<div id="embed-footer" style="margin-top: 8px; font-size: 12px; color: #72767d;"></div>
</div>
<small style="color: #666;">Cette prévisualisation est approximative. L'apparence réelle sur Discord peut varier légèrement.</small>
</div>
</div>
<script>
function formatText(text, vars) {
if (!text) return '';
return text.replace(/\{(\w+)\}/g, function(match, key) {
return vars[key] || match;
});
}
function updatePreview() {
const embedTitle = document.getElementById('embed_title').value || '{video_title}';
const embedDescription = document.getElementById('embed_description').value || '';
const embedColor = document.getElementById('embed_color_text').value || 'FF0000';
const embedAuthorName = document.getElementById('embed_author_name').value || '{channel_name}';
const embedAuthorIcon = document.getElementById('embed_author_icon').value || 'https://www.youtube.com/img/desktop/yt_1200.png';
const embedFooter = document.getElementById('embed_footer').value || '';
const embedThumbnail = document.getElementById('embed_thumbnail').checked;
const embedImage = document.getElementById('embed_image').checked;
const vars = {
video_title: 'Nouvelle vidéo de test',
channel_name: 'Ma Chaîne YouTube',
video_url: 'https://www.youtube.com/watch?v=dQw4w9WgXcQ',
video_id: 'dQw4w9WgXcQ',
thumbnail: 'https://i.ytimg.com/vi/dQw4w9WgXcQ/maxresdefault.jpg',
published_at: '2026-01-25T12:00:00Z',
is_short: false
};
document.getElementById('embed-title').textContent = formatText(embedTitle, vars);
document.getElementById('embed-title').href = vars.video_url;
document.getElementById('embed-description').textContent = formatText(embedDescription, vars);
document.getElementById('embed-author-name').textContent = formatText(embedAuthorName, vars);
document.getElementById('embed-author-icon').src = embedAuthorIcon;
document.getElementById('embed-footer').textContent = formatText(embedFooter, vars);
document.getElementById('embed-preview').style.borderLeftColor = '#' + embedColor;
if (embedThumbnail) {
document.getElementById('embed-thumbnail').src = vars.thumbnail;
document.getElementById('embed-thumbnail').style.display = 'block';
} else {
document.getElementById('embed-thumbnail').style.display = 'none';
}
if (embedImage) {
document.getElementById('embed-image').src = vars.thumbnail;
document.getElementById('embed-image').style.display = 'block';
} else {
document.getElementById('embed-image').style.display = 'none';
}
}
document.getElementById('embed_color').addEventListener('input', function(e) {
document.getElementById('embed_color_text').value = e.target.value.substring(1).toUpperCase();
updatePreview();
});
document.getElementById('embed_color_text').addEventListener('input', function(e) {
const val = e.target.value.replace(/[^0-9A-Fa-f]/g, '').substring(0, 6);
e.target.value = val;
if (val.length === 6) {
document.getElementById('embed_color').value = '#' + val;
updatePreview();
}
});
const formFields = ['embed_title', 'embed_description', 'embed_author_name', 'embed_author_icon', 'embed_footer', 'embed_thumbnail', 'embed_image'];
formFields.forEach(field => {
const el = document.getElementById(field);
if (el) {
el.addEventListener('input', updatePreview);
el.addEventListener('change', updatePreview);
}
});
updatePreview();
</script>
<p>
<strong>Variables disponibles pour l'embed :</strong>
<ul>
<li><code>{channel_name}</code> : nom de la chaîne YouTube</li>
<li><code>{video_title}</code> : titre de la vidéo</li>
<li><code>{video_url}</code> : lien vers la vidéo</li>
<li><code>{video_id}</code> : ID de la vidéo</li>
<li><code>{thumbnail}</code> : URL de la miniature</li>
<li><code>{published_at}</code> : date de publication</li>
<li><code>{is_short}</code> : True si c'est un short, False sinon</li>
</ul>
</p>
{% endblock %}

181
webapp/youtube.py Normal file
View File

@@ -0,0 +1,181 @@
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'<link rel="canonical" href="https://www\.youtube\.com/channel/([^"]{24})"', response.text)
if canonical_match:
return canonical_match.group(1)
return None
except Exception:
return None
@webapp.route("/youtube")
def openYouTube():
notifications: list[YouTubeNotification] = YouTubeNotification.query.all()
channels = bot.getAllTextChannel()
for notification in notifications:
for channel in channels:
if notification.notify_channel == channel.id:
notification.notify_channel_name = channel.name
msg = request.args.get('msg')
msg_type = request.args.get('type', 'info')
return render_template("youtube.html", notifications=notifications, channels=channels, msg=msg, msg_type=msg_type)
@webapp.route("/youtube/add", methods=['POST'])
def addYouTube():
channel_input = request.form.get('channel_id', '').strip()
channel_id = extract_channel_id(channel_input)
if not channel_id:
return redirect(url_for("openYouTube") + "?" + 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("openYouTube") + "?" + 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("openYouTube") + "?" + urlencode({'msg': "Canal Discord invalide.", 'type': 'error'}))
embed_color = request.form.get('embed_color', 'FF0000').strip().lstrip('#')
if len(embed_color) != 6:
embed_color = 'FF0000'
notification = YouTubeNotification(
enable=True,
channel_id=channel_id,
notify_channel=notify_channel,
message=request.form.get('message'),
video_type=request.form.get('video_type', 'all'),
embed_title=request.form.get('embed_title') or None,
embed_description=request.form.get('embed_description') or None,
embed_color=embed_color,
embed_footer=request.form.get('embed_footer') or None,
embed_author_name=request.form.get('embed_author_name') or None,
embed_author_icon=request.form.get('embed_author_icon') or None,
embed_thumbnail=request.form.get('embed_thumbnail') == 'on',
embed_image=request.form.get('embed_image') == 'on'
)
db.session.add(notification)
db.session.commit()
return redirect(url_for("openYouTube") + "?" + urlencode({'msg': f"Notification ajoutée avec succès pour la chaîne {channel_id}", 'type': 'success'}))
@webapp.route("/youtube/toggle/<int:id>")
def toggleYouTube(id):
notification: YouTubeNotification = YouTubeNotification.query.get_or_404(id)
notification.enable = not notification.enable
db.session.commit()
return redirect(url_for("openYouTube"))
@webapp.route("/youtube/edit/<int:id>")
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/<int:id>", 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'}))
embed_color = request.form.get('embed_color', 'FF0000').strip().lstrip('#')
if len(embed_color) != 6:
embed_color = 'FF0000'
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')
notification.embed_title = request.form.get('embed_title') or None
notification.embed_description = request.form.get('embed_description') or None
notification.embed_color = embed_color
notification.embed_footer = request.form.get('embed_footer') or None
notification.embed_author_name = request.form.get('embed_author_name') or None
notification.embed_author_icon = request.form.get('embed_author_icon') or None
notification.embed_thumbnail = request.form.get('embed_thumbnail') == 'on'
notification.embed_image = request.form.get('embed_image') == 'on'
db.session.commit()
return redirect(url_for("openYouTube") + "?" + urlencode({'msg': "Notification modifiée avec succès", 'type': 'success'}))
@webapp.route("/youtube/del/<int:id>")
def delYouTube(id):
notification = YouTubeNotification.query.get_or_404(id)
db.session.delete(notification)
db.session.commit()
return redirect(url_for("openYouTube"))