345 lines
13 KiB
Python
345 lines
13 KiB
Python
import logging
|
|
import asyncio
|
|
import xml.etree.ElementTree as ET
|
|
import requests
|
|
import discord
|
|
|
|
from database import db
|
|
from database.models import YouTubeNotification
|
|
from webapp import webapp
|
|
|
|
logger = logging.getLogger('youtube-notification')
|
|
logger.setLevel(logging.INFO)
|
|
|
|
_youtube_first_check = True
|
|
|
|
|
|
async def checkYouTubeVideos():
|
|
global _youtube_first_check
|
|
with webapp.app_context():
|
|
try:
|
|
notifications: list[YouTubeNotification] = YouTubeNotification.query.filter_by(enable=True).all()
|
|
|
|
for notification in notifications:
|
|
try:
|
|
await _checkChannelVideos(notification, is_first_check=_youtube_first_check)
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors de la vérification de la chaîne {notification.channel_id}: {e}")
|
|
db.session.rollback()
|
|
continue
|
|
|
|
if _youtube_first_check:
|
|
_youtube_first_check = False
|
|
logger.info("YouTube: première vérification terminée, notifications activées")
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors de la vérification YouTube: {e}")
|
|
db.session.rollback()
|
|
|
|
|
|
def _extract_embed_config(notification: YouTubeNotification) -> dict:
|
|
"""Extrait toutes les valeurs ORM nécessaires à l'envoi dans un dict plain Python.
|
|
Doit être appelé pendant que le contexte Flask est actif."""
|
|
return {
|
|
'notify_channel': notification.notify_channel,
|
|
'message_template': notification.message or '',
|
|
'embed_title': notification.embed_title,
|
|
'embed_description': notification.embed_description,
|
|
'embed_color': notification.embed_color or 'FF0000',
|
|
'embed_footer': notification.embed_footer,
|
|
'embed_author_name': notification.embed_author_name,
|
|
'embed_author_icon': (notification.embed_author_icon or '').strip(),
|
|
'embed_thumbnail': bool(notification.embed_thumbnail),
|
|
'embed_image': bool(notification.embed_image),
|
|
}
|
|
|
|
|
|
async def _checkChannelVideos(notification: YouTubeNotification, is_first_check: bool = False):
|
|
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
|
|
|
|
video_data = {
|
|
'title': video_title,
|
|
'url': video_url,
|
|
'published': published_at,
|
|
'channel_name': channel_name,
|
|
'thumbnail': thumbnail,
|
|
'is_short': is_short
|
|
}
|
|
|
|
if notification.video_type == 'all':
|
|
videos.append((video_id, video_data))
|
|
elif notification.video_type == 'short' and is_short:
|
|
videos.append((video_id, video_data))
|
|
elif notification.video_type == 'video' and not is_short:
|
|
videos.append((video_id, video_data))
|
|
|
|
videos.sort(key=lambda x: x[1]['published'], reverse=True)
|
|
|
|
# Enregistrer toutes les vidéos du flux dans l'historique (les doublons sont ignorés)
|
|
for vid, vdata in videos:
|
|
_save_video_history(notification.id, vid, vdata, notified=False)
|
|
|
|
if not videos:
|
|
return
|
|
|
|
latest_video_id, _ = videos[0]
|
|
if is_first_check or not notification.last_video_id:
|
|
# Au démarrage, on initialise le curseur sans annoncer l'historique.
|
|
notification.last_video_id = latest_video_id
|
|
db.session.commit()
|
|
logger.info(f"YouTube: synchronisation initiale pour {channel_id}, dernière vidéo: {latest_video_id}")
|
|
return
|
|
|
|
if latest_video_id == notification.last_video_id:
|
|
return
|
|
|
|
# Une chaîne peut publier plusieurs vidéos entre deux contrôles : le choix
|
|
# fonctionnel est d'annoncer uniquement la plus récente, jamais l'historique.
|
|
logger.info(f"Nouvelle vidéo détectée: {latest_video_id} pour la chaîne {channel_id}")
|
|
embed_config = _extract_embed_config(notification)
|
|
success = await _notifyVideo(embed_config, videos[0][1], latest_video_id)
|
|
if not success:
|
|
# Ne pas avancer le curseur : cette dernière vidéo sera réessayée au prochain cycle.
|
|
logger.warning(f"Notification échouée pour {latest_video_id}; nouvel essai au prochain contrôle")
|
|
return
|
|
|
|
_save_video_history(notification.id, latest_video_id, videos[0][1], notified=True)
|
|
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}")
|
|
db.session.rollback()
|
|
|
|
|
|
def _save_video_history(notification_id: int, video_id: str, video_data: dict, notified: bool):
|
|
"""Enregistre une vidéo dans l'historique (ne fait rien si déjà présente)."""
|
|
from database.models import YouTubeVideoHistory
|
|
try:
|
|
existing = YouTubeVideoHistory.query.filter_by(
|
|
notification_id=notification_id, video_id=video_id
|
|
).first()
|
|
if existing:
|
|
if notified and not existing.notified:
|
|
existing.notified = True
|
|
db.session.commit()
|
|
return
|
|
entry = YouTubeVideoHistory(
|
|
notification_id=notification_id,
|
|
video_id=video_id,
|
|
title=video_data.get('title', 'Sans titre'),
|
|
url=video_data.get('url', f"https://www.youtube.com/watch?v={video_id}"),
|
|
channel_name=video_data.get('channel_name', 'Inconnu'),
|
|
thumbnail=video_data.get('thumbnail'),
|
|
published_at=video_data.get('published', ''),
|
|
is_short=video_data.get('is_short', False),
|
|
notified=notified,
|
|
)
|
|
db.session.add(entry)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors de l'enregistrement de l'historique vidéo: {e}")
|
|
db.session.rollback()
|
|
|
|
|
|
async def _notifyVideo(embed_config: dict, video_data: dict, video_id: str) -> bool:
|
|
"""Envoie la notification Discord. Retourne True si l'envoi a réussi."""
|
|
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)
|
|
|
|
message_template = embed_config.get('message_template', '')
|
|
try:
|
|
message = message_template.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, AttributeError, ValueError) as e:
|
|
logger.error(f"Erreur de formatage du message: {e}")
|
|
message = f"🎥 Nouvelle vidéo de {channel_name}: [{video_title}]({video_url})"
|
|
|
|
logger.info(f"Envoi de notification YouTube: {message}")
|
|
return await _sendMessage(embed_config, 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}")
|
|
return False
|
|
|
|
|
|
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(embed_config: dict, message: str, video_url: str, thumbnail: str, video_title: str, channel_name: str, video_id: str, published_at: str, is_short: bool) -> bool:
|
|
"""Envoie le message Discord. Retourne True si l'envoi a réussi."""
|
|
from discordbot import bot
|
|
try:
|
|
channel_id = int(embed_config['notify_channel'])
|
|
discord_channel = bot.get_channel(channel_id)
|
|
if not discord_channel:
|
|
# Le salon peut ne pas être présent dans le cache local après une reconnexion.
|
|
discord_channel = await bot.fetch_channel(channel_id)
|
|
if not discord_channel:
|
|
logger.error(f"Canal Discord {channel_id} introuvable")
|
|
return False
|
|
|
|
embed_title_text = _format_embed_text(embed_config['embed_title'], channel_name, video_title, video_url, video_id, thumbnail, published_at, is_short) if embed_config['embed_title'] else video_title
|
|
embed_description = _format_embed_text(embed_config['embed_description'], channel_name, video_title, video_url, video_id, thumbnail, published_at, is_short) if embed_config['embed_description'] else None
|
|
|
|
try:
|
|
embed_color = int(embed_config['embed_color'], 16)
|
|
except ValueError:
|
|
embed_color = 0xFF0000
|
|
|
|
embed = discord.Embed(
|
|
title=embed_title_text,
|
|
url=video_url,
|
|
color=embed_color
|
|
)
|
|
|
|
if embed_description:
|
|
embed.description = embed_description
|
|
|
|
author_name = _format_embed_text(embed_config['embed_author_name'], channel_name, video_title, video_url, video_id, thumbnail, published_at, is_short) if embed_config['embed_author_name'] else channel_name
|
|
author_icon_raw = embed_config['embed_author_icon']
|
|
author_icon = author_icon_raw if author_icon_raw.startswith(("http://", "https://")) else "https://www.youtube.com/img/desktop/yt_1200.png"
|
|
embed.set_author(name=author_name, icon_url=author_icon)
|
|
|
|
if embed_config['embed_thumbnail'] and thumbnail:
|
|
embed.set_thumbnail(url=thumbnail)
|
|
|
|
if embed_config['embed_image'] and thumbnail:
|
|
embed.set_image(url=thumbnail)
|
|
|
|
if embed_config['embed_footer']:
|
|
footer_text = _format_embed_text(embed_config['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")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors de l'envoi du message Discord: {e}")
|
|
return False
|
|
|
|
|
|
async def _send_video_notification_async(history_id: int) -> tuple[bool, str]:
|
|
"""Force l'envoi d'une notification pour une vidéo de l'historique. Retourne (succès, message)."""
|
|
from database.models import YouTubeVideoHistory
|
|
with webapp.app_context():
|
|
history = YouTubeVideoHistory.query.get(history_id)
|
|
if not history:
|
|
return (False, "Vidéo introuvable dans l'historique.")
|
|
|
|
notification = YouTubeNotification.query.get(history.notification_id)
|
|
if not notification:
|
|
return (False, "Notification YouTube associée introuvable.")
|
|
|
|
embed_config = _extract_embed_config(notification)
|
|
video_data = {
|
|
'title': history.title or 'Sans titre',
|
|
'url': history.url or f"https://www.youtube.com/watch?v={history.video_id}",
|
|
'channel_name': history.channel_name or 'Inconnu',
|
|
'thumbnail': history.thumbnail or '',
|
|
'published': history.published_at or '',
|
|
'is_short': history.is_short,
|
|
}
|
|
|
|
success = await _notifyVideo(embed_config, video_data, history.video_id)
|
|
if success:
|
|
history.notified = True
|
|
db.session.commit()
|
|
return (True, "Notification envoyée sur Discord.")
|
|
else:
|
|
db.session.rollback()
|
|
return (False, "Échec de l'envoi sur Discord.")
|
|
|
|
|
|
def send_video_notification_sync(history_id: int) -> tuple[bool, str]:
|
|
"""Appel synchrone pour forcer une notification (depuis la webapp)."""
|
|
from discordbot import bot
|
|
try:
|
|
future = asyncio.run_coroutine_threadsafe(
|
|
_send_video_notification_async(history_id),
|
|
bot.loop,
|
|
)
|
|
return future.result(timeout=15)
|
|
except Exception as e:
|
|
logger.error(f"send_video_notification_sync: {e}")
|
|
return (False, str(e))
|