finish basic functionality, set up ghcr.io

.github/workflows/code-quality.yml

@@ -1,6 +1,6 @@
 name: Code Quality
 
-on: [push, pull_request]
+on: [push, pull_request, workflow_call]
 
 jobs:
   lint:

.github/workflows/docker-publish.yml

@@ -0,0 +1,36 @@
+name: 'Publish release image to ghcr.io'
+on:
+  # publish on releases, e.g. v2.1.13 (image tagged as "2.1.13")
+  # NB: "v" prefix is removed
+  release:
+    types:
+      - published
+
+  # publish "latest" tag on pushes to the main branch
+  push:
+    branches:
+      - main
+
+jobs:
+  code_quality:
+    uses: './.github/workflows/code-quality.yml'
+
+  build_dockerfile:
+    uses: './.github/workflows/docker.yml'
+
+  publish_release:
+    runs-on: ubuntu-latest
+    permissions:
+      packages: write
+      contents: read
+    # TODO: Re-enable code_quality after we fix ruff etc.
+    # needs: [code_quality, build_dockerfile]
+    needs: build_dockerfile
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Build and publish a Docker image for ${{ github.repository }}
+        uses: macbre/push-to-ghcr@master
+        with:
+          image_name: ${{ github.repository }}
+          github_token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/docker.yml

@@ -1,6 +1,6 @@
 name: Docker
 
-on: [push, pull_request]
+on: [push, pull_request, workflow_call]
 
 jobs:
   lint:

src/minecraft/__init__.py

@@ -1,5 +1,6 @@
 """Create resource packs for Minecraft and discs.json for Jukebox Extended Reborn."""
 
+from .jext import populate_discs_json  # noqa:F401
 from .resource_pack import ResourcePack  # noqa: F401
 
 # This package is versioned independently from the parent project.

src/minecraft/jext.py

@@ -0,0 +1,26 @@
+import json
+
+
+def populate_discs_json(
+    *,
+    template_file: str,
+    title: str,
+    author: str,
+    duration: int,
+    lores: list[str] = None,
+) -> dict:
+    with open(template_file, encoding="utf-8") as template_file:
+        discs = json.load(template_file)
+
+    discs[0]["title"] = str(title)
+    discs[0]["author"] = str(author)
+    discs[0]["duration"] = int(duration)
+
+    # `lores` is optional, custom lores can be specified in the template file
+    if lores:
+        if len(lores < 3):
+            discs[0]["lores"] = lores
+        else:
+            raise ValueError(f"lores array has too many items! ({len(lores)} > 2)")
+
+    return discs

src/rss2jext.py

@@ -1,17 +1,16 @@
 import argparse
 import json
+import math
 import os
 
 import requests
 from dotenv import load_dotenv
 from ffmpeg import FFmpeg, Progress
 from mutagen.oggvorbis import OggVorbis
-from rss_parser import Parser
-from rss_parser.models.item import Item
-from rss_parser.models.types.tag import Tag
 
-from minecraft import ResourcePack
+from minecraft import ResourcePack, populate_discs_json
 from pterodactyl import Client
+from rss_helper import RSSHelper
 
 __version__ = "1.0.0"
 
@@ -29,46 +28,6 @@ def load_servers() -> dict:
         return json.load(servers_json)
 
 
-def rss_feed(feed_url: str):
-    print(f"[RSS] Downloading feed from {feed_url}...")
-
-    feed_req = requests.get(feed_url, timeout=1000, headers={"User-Agent": USER_AGENT})
-    feed_req.raise_for_status()
-
-    return Parser.parse(feed_req.text)
-
-
-def rss_latest_episode(rss) -> Tag[Item]:
-    print(f"[RSS] Feed language: {rss.channel.language}")
-    print(f"[RSS] Feed version: {rss.version}")
-
-    rss_items = rss.channel.items
-    # The feed is ordered by date ascending so the newest episodes are at the end
-    # of rss_items
-    rss_items.reverse()
-
-    i = 0
-
-    # TODO: Check the number of items in rss_items before entering this loop
-    while True:
-        i = i + 1
-
-        latest_rss_item = rss_items.pop()
-        title = latest_rss_item.title.lower()
-
-        # TODO: Make this filter customizable
-        if not title.startswith("teaser"):
-            return latest_rss_item
-
-        # HACK: Find a better way of handling this edge case
-        if i > 5:  # abort after 5 iterations so we're not here all day
-            raise RuntimeError("Could not find a non-teaser episode after 5 tries.")
-
-
-def extract_mp3_url(episode: Item) -> str:
-    return episode.enclosure.attributes["url"]
-
-
 # TODO: Maybe add an extra argument to this that lets us specify where to save the file
 def download_mp3(url: str) -> str:
     # TODO: Don't hardcode timeout
@@ -153,11 +112,6 @@ def build_packs(versions: list, *, pack_description: str):
         print(f"[minecraft] Resource pack built: {rp_filename}")
 
 
-def build_jext_config(*, oggfile, rss_feed):
-    # TODO: write this. Taco Bell time yum!!!
-    pass
-
-
 def main() -> None:
     load_dotenv()
 
@@ -169,15 +123,15 @@ def main() -> None:
     parser.add_argument("--skip-encode", action="store_true")
     args = parser.parse_args()
 
-    feed = rss_feed(os.getenv("RSS_URL"))
-    episode = rss_latest_episode(feed)
-
     ptero_servers = load_servers()
 
-    print(f"[RSS] Latest episode Title: {episode.title}")
-    print(f"[RSS] Latest episode GUID: {episode.guid}")
+    rss_feed = RSSHelper(os.getenv("RSS_URL"), user_agent=USER_AGENT)
+    latest_episode = rss_feed.latest_episode()
+
+    print(f"[RSS] Latest episode Title: {latest_episode.title}")
+    print(f"[RSS] Latest episode GUID: {latest_episode.guid}")
 
-    ep_url = extract_mp3_url(episode=episode)
+    ep_url = rss_feed.episode_file_url(rss_feed.latest_episode())
 
     print(f"[RSS] Latest episode URL: {ep_url}")
 
@@ -190,14 +144,32 @@ def main() -> None:
 
     if args.skip_encode:
         print("[ffmpeg] Skipping mp3 -> ogg encoding due to --skip-encode")
+        # Try to assign a default value if we skip encoding
+        ogg_file = os.path.join(__data_dir__, "tmp", "episode.ogg")
     else:
         print(f"[ffmpeg] encoding {mp3_file} -> ogg")
         ogg_file = mp3_to_ogg(mp3_file)
         print(f"[ffmpeg] done: {ogg_file}")
 
-    build_packs([15, 18], pack_description=episode.title)
+    build_packs([15, 18], pack_description=latest_episode.title)
 
     audio_duration = OggVorbis(ogg_file).info.length
+    audio_duration = math.ceil(audio_duration)
+
+    print(f"[ogg] file has a duration of {audio_duration} seconds")
+
+    discs_json = populate_discs_json(
+        template_file=os.path.join(__data_dir__, "templates", "discs.json"),
+        title=latest_episode.title,
+        author=rss_feed.feed().channel.content.title,
+        duration=audio_duration,
+    )
+
+    print("[discs.json] writing discs.json")
+    with open(
+        os.path.join(__data_dir__, "out", "discs.json"), "w", encoding="utf-8"
+    ) as discs_file:
+        json.dump(discs_json, discs_file, allow_nan=False, indent=4)
 
 
 if __name__ == "__main__":

src/rss_helper/__init__.py

@@ -0,0 +1,3 @@
+"""Simple helper module for managing RSS feeds."""
+
+from .rss_helper import RSSHelper  # noqa: F401

src/rss_helper/rss_helper.py

@@ -0,0 +1,72 @@
+"""Simple helper class for managing RSS feeds."""
+import requests
+from rss_parser import Parser
+from rss_parser.models.item import Item
+from rss_parser.models.types import Tag
+
+
+class RSSHelper:
+
+    """Simple helper class for managing RSS feeds."""
+
+    __feed_url: str = ""
+    __feed = None
+    __user_agent = None
+
+    def __init__(self, url: str, *, user_agent: str):
+        """Initialize internal feed URL, feed object, and user-agent."""
+        self.__feed_url = url
+        self.__user_agent = user_agent
+
+        self.__feed = self.from_url(self.__feed_url)
+
+    def feed(self):
+        return self.__feed
+
+    def from_url(self, url: str):
+        """Create an RSS object from a feed URL."""
+        print(f"[RSS] Downloading feed from {url}...")
+
+        feed_req = requests.get(
+            url, timeout=1000, headers={"User-Agent": self.__user_agent}
+        )
+
+        feed_req.raise_for_status()
+
+        return Parser.parse(feed_req.text)
+
+    def latest_episode(self, *, feed=None) -> Tag[Item]:
+        """Get the latest episode from an RSS feed."""
+        if not feed:
+            feed = self.__feed
+
+        print(f"[RSS] Feed language: {feed.channel.language}")
+        print(f"[RSS] Feed version: {feed.version}")
+
+        feed_items = feed.channel.items
+
+        # The feed is ordered by date ascending so the newest episodes are at
+        # the end of rss_items
+        feed_items.reverse()
+
+        i = 0
+
+        # TODO: Check the number of items in rss_items before entering this loop
+        while True:
+            i = i + 1
+
+            latest_rss_item = feed_items.pop()
+
+            title = latest_rss_item.title.lower()
+
+            # TODO: Make this filter customizable
+            if not title.startswith("teaser"):
+                return latest_rss_item
+
+            # HACK: Find a better way of handling this edge case
+            if i > 5:  # abort after 5 iterations so we're not here all day
+                raise RuntimeError("Could not find a non-teaser episode after 5 tries.")
+
+    def episode_file_url(self, episode) -> str:
+        """Get the URL to download the audio file for an episode."""
+        return episode.enclosure.attributes["url"]