Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

cache folders are now created in a proper temp folder location #22

Merged
merged 2 commits into from
Jan 20, 2024
Merged

cache folders are now created in a proper temp folder location #22

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4bd1368
use TemporaryDirectory as caching folder
alexhartm Jan 20, 2024
26b20c2
debug msg
alexhartm Jan 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 50 additions & 52 deletions toniepodcastsync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
"""The Tonie Podcast Sync API."""
import logging
import platform
import shutil
import subprocess
import tempfile
from io import BytesIO
from pathlib import Path

Expand Down Expand Up @@ -73,53 +73,55 @@ def sync_podcast_to_tonie(self, podcast: Podcast, tonie_id: str, max_minutes: in
tonie_id (str): The id of the tonie
max_minutes (int, optional): The maximum time of podcasts in length. Defaults to 90.
"""
if tonie_id not in self.__tonieDict:
msg = f"ERROR: Cannot find tonie with ID {tonie_id}"
log.error(msg)
console.print(msg, style="red")
return
if len(podcast.epList) == 0:
msg = (
f"ERROR: Cannot find any episodes at all for podcast '{podcast.title}'"
f"to put on tonie with ID {tonie_id}"
)
log.warning(msg)
console.print(msg, style="orange")
return
if not self.__is_tonie_empty(tonie_id):
# check if new feed has newer epsiodes than tonie
latest_episode_feed = self.__generate_chapter_title(podcast.epList[0])
latest_episode_tonie = self.__tonieDict[tonie_id].chapters[0].title
if latest_episode_tonie == latest_episode_feed:
msg = f"Podcast '{podcast.title}' has no new episodes, latest episode is '{latest_episode_tonie}'"
log.info(msg)
console.print(msg)
with tempfile.TemporaryDirectory() as podcast_cache_directory:
self.podcast_cache_directory = Path(podcast_cache_directory)
msg = f"DEBUG: cache path is {self.podcast_cache_directory}"
log.debug(msg)
if tonie_id not in self.__tonieDict:
msg = f"ERROR: Cannot find tonie with ID {tonie_id}"
log.error(msg)
console.print(msg, style="red")
return
else:
log.info("### tonie is empty")
# add new episodes to tonie
self.__wipe_tonie(tonie_id)
cached_episodes = self.__cache_podcast_episodes(podcast, max_minutes)

for e in track(
cached_episodes,
description=(
f"{podcast.title}: transferring {len(cached_episodes)} episodes"
f" to {self.__tonieDict[tonie_id].name}"
),
total=len(cached_episodes),
transient=True,
refresh_per_second=2,
):
self.__upload_episode(e, tonie_id)

episode_info = [f"{episode.title} ({episode.published})" for episode in cached_episodes]
console.print(
f"{podcast.title}: Successfully uploaded {episode_info} to "
f"{self.__tonieDict[tonie_id].name} ({self.__tonieDict[tonie_id].id})",
)

self.__cleanup_cache()
if len(podcast.epList) == 0:
msg = (
f"ERROR: Cannot find any episodes at all for podcast '{podcast.title}'"
f"to put on tonie with ID {tonie_id}"
)
log.warning(msg)
console.print(msg, style="orange")
return
if not self.__is_tonie_empty(tonie_id):
# check if new feed has newer epsiodes than tonie
latest_episode_feed = self.__generate_chapter_title(podcast.epList[0])
latest_episode_tonie = self.__tonieDict[tonie_id].chapters[0].title
if latest_episode_tonie == latest_episode_feed:
msg = f"Podcast '{podcast.title}' has no new episodes, latest episode is '{latest_episode_tonie}'"
log.info(msg)
console.print(msg)
return
else:
log.info("### tonie is empty")
# add new episodes to tonie
self.__wipe_tonie(tonie_id)
cached_episodes = self.__cache_podcast_episodes(podcast, max_minutes)

for e in track(
cached_episodes,
description=(
f"{podcast.title}: transferring {len(cached_episodes)} episodes"
f" to {self.__tonieDict[tonie_id].name}"
),
total=len(cached_episodes),
transient=True,
refresh_per_second=2,
):
self.__upload_episode(e, tonie_id)

episode_info = [f"{episode.title} ({episode.published})" for episode in cached_episodes]
console.print(
f"{podcast.title}: Successfully uploaded {episode_info} to "
f"{self.__tonieDict[tonie_id].name} ({self.__tonieDict[tonie_id].id})",
)

def __upload_episode(self, ep: Episode, tonie_id: str) -> None:
# upload a given episode to a creative tonie
Expand Down Expand Up @@ -177,7 +179,7 @@ def __cache_podcast_episodes(self, podcast: Podcast, max_minutes: int = MAXIMUM_
def __cache_episode(self, ep: Episode) -> bool:
# local download of a single episode into a subfolder
# file name is build according to __generateFilename
podcast_path = Path("podcasts") / sanitize_filepath(ep.podcast)
podcast_path = self.podcast_cache_directory / sanitize_filepath(ep.podcast)
podcast_path.mkdir(parents=True, exist_ok=True)

fname = podcast_path / self.__generate_filename(ep)
Expand Down Expand Up @@ -205,10 +207,6 @@ def __generate_filename(self, ep: Episode) -> str:
# generates canonical filename for local episode cache
return sanitize_filename(f"{ep.published} {ep.title}.mp3")

def __cleanup_cache(self) -> None:
console.print("Cleanup the cache folder.")
shutil.rmtree(Path("podcasts"), ignore_errors=True)

def __generate_chapter_title(self, ep: Episode) -> str:
# generate chapter title used when writing on tonie
return ep.title + " (" + ep.published + ")"
Expand Down
Loading