mirror of
https://github.com/skylanix/MamieHenriette.git
synced 2026-02-06 06:40:35 +01:00
226 lines
7.8 KiB
Python
226 lines
7.8 KiB
Python
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}")
|