Upgraded modules, fixed error handling because Ruff complained

This commit is contained in:
florian committed 2026-10-21 09:08:48 +02:00
1 parent afb8c6059a
commit 2e2c82ad22
6 files changed
+112 -89

No files matched your search

+16 -13
View File
@@ -1,12 +1,15 @@
import requests
import yt_dlp
import os import os
import time import time
import requests
import yt_dlp
from dotenv import load_dotenv from dotenv import load_dotenv
from ssh_helper import upload_via_sftp, send_notification_via_ssh
from youtube_handler import return_download_options, check_for_sponsorblock_segments
from simple_logger_handler import setup_logger from simple_logger_handler import setup_logger
from rss_feed_handler import grab_latest_chapter_information, EpisodeData
from rss_feed_handler import EpisodeData, FeedError, grab_latest_chapter_information
from ssh_helper import send_notification_via_ssh, upload_via_sftp
from youtube_handler import check_for_sponsorblock_segments, return_download_options
logger = setup_logger(__name__) logger = setup_logger(__name__)
load_dotenv() load_dotenv()
@@ -84,8 +87,8 @@ def check_until_new_episode_gets_released() -> tuple[EpisodeData | None, str | N
try: try:
episode_data = grab_latest_chapter_information("UCC3ehuUksTyQ7bbjGntmx3Q") episode_data = grab_latest_chapter_information("UCC3ehuUksTyQ7bbjGntmx3Q")
except Exception as e: except FeedError as e:
logger.warning(f"[EpisodeCheck] Failed to fetch latest video data: {e}, retrying in 1 hour.") logger.warning(f"[EpisodeCheck] Failed to fetch latest video data: {e!r}, retrying in 1 hour.")
time.sleep(CHECK_INTERVAL_HOURS * 3600) time.sleep(CHECK_INTERVAL_HOURS * 3600)
continue continue
@@ -144,8 +147,8 @@ def download_episode() -> None:
return return
logger.info("[App] New episode found") logger.info("[App] New episode found")
except Exception as e: except Exception:
logger.error(f"[App] Failed to fetch new episode info: {e}", exc_info=True) logger.exception("[App] Failed to fetch new episode info")
return return
try: try:
@@ -165,8 +168,8 @@ def download_episode() -> None:
try: try:
options = return_download_options(episode_data) options = return_download_options(episode_data)
except Exception as e: except Exception:
logger.error(f"[App] Failed to generate download options: {e}", exc_info=True) logger.exception("[App] Failed to generate download options")
return return
logger.info("[App] Downloading new episode") logger.info("[App] Downloading new episode")
@@ -174,8 +177,8 @@ def download_episode() -> None:
with yt_dlp.YoutubeDL(options) as episode: with yt_dlp.YoutubeDL(options) as episode:
episode.download(episode_data.episode_link) episode.download(episode_data.episode_link)
logger.debug("[App] Download completed successfully") logger.debug("[App] Download completed successfully")
except Exception as e: except Exception:
logger.error(f"[App] Failed to download episode: {e}", exc_info=True) logger.exception("[App] Failed to download episode")
return return
logger.info("[App] Uploading episode via SFTP") logger.info("[App] Uploading episode via SFTP")
+1
View File
@@ -1,5 +1,6 @@
import re import re
def return_string_as_html(input_text): def return_string_as_html(input_text):
string_without_ads="" string_without_ads=""
for line in input_text.splitlines(): for line in input_text.splitlines():
+21 -19
View File
@@ -1,23 +1,25 @@
--extra-index-url https://git.gansejunge.com/api/packages/notifier/pypi/simple/ --extra-index-url https://git.gansejunge.com/api/packages/notifier/pypi/simple/
bcrypt==5.0.0 bcrypt==5.0.0
Brotli==1.1.0 brotli==1.2.0
certifi==2026.10.19 certifi==2026.10.19
cffi==2.0.0 cffi==2.1.1
charset-normalizer==3.4.3 charset-normalizer==3.5.2
cryptography==46.0.2 cryptography==50.0.2
deno==2.9.7
dotenv==0.9.9 dotenv==0.9.9
feedparser==6.0.12 feedparser==6.0.14
idna==3.10 feedparser-sgmllib==2.1.0
invoke==2.2.0 idna==3.20
mutagen==1.47.0 invoke==3.0.3
paramiko==4.0.0 mutagen==1.48.1
pycparser==2.23 paramiko==5.1.0
pycryptodomex==3.23.0 pycparser==3.11
PyNaCl==1.6.0 pycryptodomex==3.24.1
python-dotenv==1.1.1 PyNaCl==1.6.2
requests==2.32.5 python-dotenv==1.2.4
sgmllib3k==1.0.0 python-json-logger==4.2.0
simple-logger-handler==0.1.0 requests==2.34.2
sponsorblock.py==0.2.3 simple-logger-handler==0.2.1
urllib3==2.5.0 sponsorblock.py==1.0.0
websockets==15.0.1 urllib3==2.8.0
websockets==17.3
+56 -38
View File
@@ -1,11 +1,22 @@
import feedparser
from simple_logger_handler import setup_logger
import time import time
from urllib.error import URLError
from typing import Optional
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime
import feedparser
from simple_logger_handler import setup_logger
class FeedError(Exception):
"""Base class for all errors raised while fetching or reading the feed."""
class FeedUnavailableError(FeedError):
"""The feed could not be fetched (network failure or HTTP error status)."""
class FeedParseError(FeedError):
"""The feed was fetched but its content is empty, malformed or unexpected."""
@dataclass @dataclass
class EpisodeData: class EpisodeData:
episode_date: str episode_date: str
@@ -17,12 +28,24 @@ class EpisodeData:
logger = setup_logger(__name__) logger = setup_logger(__name__)
def grab_latest_chapter_information(id: str, max_retries: int = 3) -> Optional[EpisodeData]: def fetch_feed(url: str) -> feedparser.FeedParserDict:
feed = feedparser.parse(url)
status = feed.get("status")
if status is None:
raise FeedUnavailableError(f"No HTTP response: {feed.get('bozo_exception')}")
if status >= 400:
raise FeedUnavailableError(f"HTTP {status}")
if not feed.entries:
raise FeedParseError(f"No entries in feed (bozo={feed.get('bozo_exception')})")
return feed
def grab_latest_chapter_information(channel_id: str, max_retries: int = 3) -> EpisodeData:
""" """
Fetches the latest episodes information from a Youtube RSS feed, with retries on network-related errors. Fetches the latest episodes information from a Youtube RSS feed, with retries on network-related errors.
Parameters: Parameters:
id: Youtube channel ID as a string. channel_id: Youtube channel ID as a string.
max_retries: Number of retry attempts if fetching the feed fails due to network issues. max_retries: Number of retry attempts if fetching the feed fails due to network issues.
Returns: Returns:
@@ -36,50 +59,45 @@ def grab_latest_chapter_information(id: str, max_retries: int = 3) -> Optional[E
Returns None if the feed has no entries or all retries are exhausted. Returns None if the feed has no entries or all retries are exhausted.
Raises: Raises:
ValueError: If the feed has no entries. FeedUnavailableError: Feed could not be fetched after all retries.
Other network-related exceptions: If fetching fails after retries. FeedParseError: Feed was fetched but is empty or has unexpected content.
""" """
rss_feed_url = f"https://www.youtube.com/feeds/videos.xml?channel_id={id}" rss_feed_url = f"https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"
attempt = 1 attempt = 1
while attempt <= max_retries: while attempt <= max_retries:
logger.debug(f"[Feed] Parsing feed URL: {rss_feed_url} (attempt {attempt}/{max_retries})") logger.debug(f"[Feed] Parsing feed URL: {rss_feed_url} (attempt {attempt}/{max_retries})")
try: try:
feed = feedparser.parse(rss_feed_url) feed = fetch_feed(rss_feed_url)
break
if not feed.entries: except FeedUnavailableError as e:
logger.warning(f"[Feed] No entries found for feed {id}") logger.warning(f"[Feed] Attempt {attempt}/{max_retries} failed for {channel_id}: {e}")
return None
latest_chapter_data = feed["entries"][0]
episode_link = latest_chapter_data["link"]
episode_title = latest_chapter_data["title"]
episode_description = latest_chapter_data["summary"]
episode_date = latest_chapter_data["published"]
episode_date = datetime.fromisoformat(episode_date).date().isoformat()
episode_ytid = latest_chapter_data["yt_videoid"]
logger.info(f"[Feed] Latest episode '{episode_title}': {episode_link}")
logger.debug(f"[Feed] Latest episode '{episode_title}' (YouTubeId {episode_ytid}): {episode_link} -> {episode_description}")
return EpisodeData(
episode_date=episode_date,
episode_description=episode_description,
episode_link=episode_link,
episode_number="",
episode_title=episode_title,
episode_ytid=episode_ytid
)
except (URLError, OSError) as e:
logger.warning(f"[Feed] Network error on attempt {attempt} for feed {id}: {e}")
if attempt == max_retries: if attempt == max_retries:
logger.error(f"[Feed] All {max_retries} attempts failed for feed {id}") logger.error(f"[Feed] All {max_retries} attempts failed for feed {channel_id}")
return None raise
backoff = 2 ** (attempt - 1) backoff = 2 ** (attempt - 1)
logger.debug(f"[Feed] Retrying in {backoff} seconds...") logger.debug(f"[Feed] Retrying in {backoff} seconds...")
time.sleep(backoff) time.sleep(backoff)
attempt += 1 attempt += 1
latest = feed.entries[0]
try:
episode_date = datetime.fromisoformat(latest["published"]).date().isoformat()
episode = EpisodeData(
episode_date=episode_date,
episode_description=latest["summary"],
episode_link=latest["link"],
episode_number="",
episode_title=latest["title"],
episode_ytid=latest["yt_videoid"],
)
except (KeyError, ValueError) as e:
raise FeedParseError(f"Unexpected entry format for {channel_id}: {e!r}") from e
logger.info(f"[Feed] Latest episode '{episode.episode_title}': {episode.episode_link}")
logger.debug(f"[Feed] Latest episode '{episode.episode_title}' (YouTubeId {episode.episode_ytid}): {episode.episode_link} -> {episode.episode_description}")
return episode
if __name__ == "__main__": if __name__ == "__main__":
print(grab_latest_chapter_information("UCC3ehuUksTyQ7bbjGntmx3Q")) print(grab_latest_chapter_information("UCC3ehuUksTyQ7bbjGntmx3Q"))
+12 -11
View File
@@ -1,10 +1,11 @@
import paramiko
import os import os
from dotenv import load_dotenv
from json import dumps
from simple_logger_handler import setup_logger
import time
import shlex import shlex
import time
from json import dumps
import paramiko
from dotenv import load_dotenv
from simple_logger_handler import setup_logger
logger = setup_logger(__name__) logger = setup_logger(__name__)
load_dotenv() load_dotenv()
@@ -78,8 +79,8 @@ def create_ssh_client(hostname: str, port: int, username: str, keyfile: str)-> p
ssh.connect(hostname=hostname, username=username, port=port, pkey=pkey) ssh.connect(hostname=hostname, username=username, port=port, pkey=pkey)
logger.debug("[SSH] SSH connection established successfully") logger.debug("[SSH] SSH connection established successfully")
return ssh return ssh
except Exception as e: except Exception:
logger.error(f"[SSH] SSH connection failed: {e}", exc_info=True) logger.exception("[SSH] SSH connection failed")
raise raise
@@ -110,8 +111,8 @@ def upload_via_sftp(filename) -> None:
sftp.close() sftp.close()
transport.close() transport.close()
logger.info(f"[SFTP] File '{filename}' uploaded successfully") logger.info(f"[SFTP] File '{filename}' uploaded successfully")
except Exception as e: except Exception:
logger.error(f"[SFTP] SFTP upload failed for '{filename}': {e}", exc_info=True) logger.exception(f"[SFTP] SFTP upload failed for '{filename}'")
raise raise
@@ -171,8 +172,8 @@ def send_notification_via_ssh(notification_title, notification_info) -> None:
logger.warning(f"[Notification] Notification command exited with {exit_status}") logger.warning(f"[Notification] Notification command exited with {exit_status}")
logger.warning(f"[Notification] Error: {error_output}") logger.warning(f"[Notification] Error: {error_output}")
logger.warning(f"[Notification] Response: {response_output}") logger.warning(f"[Notification] Response: {response_output}")
except Exception as e: except Exception:
logger.error(f"[Notification] Failed to send SSH notification: {e}", exc_info=True) logger.exception("[Notification] Failed to send SSH notification")
raise raise
finally: finally:
if ssh: if ssh:
+6 -8
View File
@@ -1,13 +1,11 @@
import yt_dlp
import datetime
import contextlib
from dotenv import load_dotenv
import os
from helper import return_string_as_html
from simple_logger_handler import setup_logger
import json import json
import sponsorblock as sb import os
import sponsorblock as sb
from dotenv import load_dotenv
from simple_logger_handler import setup_logger
from helper import return_string_as_html
logger = setup_logger(__name__) logger = setup_logger(__name__)
load_dotenv() load_dotenv()