Added Perun script

src/perun/README.md

@@ -0,0 +1,18 @@
+# Perun
+
+
+Youtube blocks a lot of server IPs so running this locally is just easier, expects the following environment variables in a .env file:
+
+REMOTE_HOSTNAME
+
+REMOTE_PATH
+
+BACKEND_API_URL
+
+BACKEND_API_KEY
+
+YOUTUBE_CHANNEL_URL
+
+PODCAST_AUTHORIZATION_TOKEN
+
+PODCAST_API_URL

src/perun/get_episode.py

@@ -0,0 +1,66 @@
+import requests
+import yt_dlp
+import os
+from dotenv import load_dotenv
+from helper import log_message
+from ssh_helper import upload_via_sftp, send_notification_via_ssh
+from youtube_handler import get_url_for_latest_video, get_youtube_data, return_download_options
+
+
+load_dotenv()
+PODCAST_AUTHORIZATION_TOKEN = os.getenv("PODCAST_AUTHORIZATION_TOKEN")
+PODCAST_API_URL =  os.getenv("PODCAST_API_URL")
+
+
+def get_audiobookshelf_data()->tuple[int | None, str | None]:
+	headers = {"Authorization": f"Bearer {PODCAST_AUTHORIZATION_TOKEN}"}
+
+	try:
+		response = requests.get(PODCAST_API_URL, headers=headers)
+		response.raise_for_status()
+		result = response.json()
+		audiobookshelf_track = result["media"]["episodes"][-1]["audioFile"]["metaTags"]["tagTrack"]
+		audiobookshelf_title = result["media"]["episodes"][-1]["audioFile"]["metaTags"]["tagTitle"]
+		return audiobookshelf_track, audiobookshelf_title
+
+	except requests.RequestException as e:
+		log_message(f"Failed to fetch data: {e}")
+		return None
+	
+
+def download_episode():
+	log_message("Starting Perun")
+
+	audiobookshelf_track, audiobookshelf_title = get_audiobookshelf_data()
+	if audiobookshelf_track is None or audiobookshelf_title is None:
+		log_message("Unable to fetch Audiobookshelf data. Exiting.")
+		return
+	
+	episode_url = get_url_for_latest_video()
+	episode_info = get_youtube_data(episode_url)
+	log_message(f"Latest episode: {episode_info['title']}")
+
+	if audiobookshelf_title != episode_info["title"]: 
+		log_message("New Episode found") 
+ 
+		track = str(int(audiobookshelf_track) + 1).zfill(4)
+		options = return_download_options(episode_info,track)
+		
+		log_message("Downloading episode")
+		try:
+			with yt_dlp.YoutubeDL(options) as episode:
+				episode.download(episode_url)
+		except Exception as e:
+			log_message(f"Failed to download episode: {e}")
+			return
+		
+		log_message("Uploading episode")
+		upload_via_sftp(f"perun-{episode_info['date']}.mp3")
+		log_message("Finished uploading, sending notification")
+		send_notification_via_ssh(f"Perun episode {track} has been released",episode_info["title"])
+		log_message("Finished")
+	else:
+		log_message("No new episode found, exiting...")
+
+if __name__ == "__main__":
+	download_episode()

src/perun/grabEpisode.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+
+cd "$(dirname "$0")"
+
+if [ ! -d ".venv" ]; then
+	echo "Creating virtual environment..."
+	python3 -m venv .venv
+fi
+
+source .venv/bin/activate
+
+pip install --upgrade pip
+pip install --upgrade yt-dlp[default]
+pip install -r requirements.txt
+
+python get_episode.py

src/perun/helper.py

@@ -0,0 +1,15 @@
+import re
+import datetime
+
+def return_string_as_html(input_text):
+	string_without_ads=""
+	for line in input_text.splitlines():
+		line = re.sub(r'(https?://[^\s]+)', r'<a href="\1">\1</a>', line)
+		if not "Sponsored" in line:
+			string_without_ads+=line+"\n"
+	return("<p>"+string_without_ads.replace("\n\n", "</p>\n<p>").replace("\n", "<br>")+"</p>")
+
+def log_message(message):
+	timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+	print(f"[{timestamp}] {message}")
+	return(f"[{timestamp}] {message}\n")

src/perun/requirements.txt

@@ -0,0 +1,18 @@
+bcrypt==5.0.0
+Brotli==1.1.0
+certifi==2025.10.5
+cffi==2.0.0
+charset-normalizer==3.4.3
+cryptography==46.0.2
+dotenv==0.9.9
+idna==3.10
+invoke==2.2.0
+mutagen==1.47.0
+paramiko==4.0.0
+pycparser==2.23
+pycryptodomex==3.23.0
+PyNaCl==1.6.0
+python-dotenv==1.1.1
+requests==2.32.5
+urllib3==2.5.0
+websockets==15.0.1

src/perun/ssh_helper.py

@@ -0,0 +1,77 @@
+import paramiko
+import os
+from dotenv import load_dotenv
+from json import dumps
+
+load_dotenv()
+REMOTE_HOSTNAME = os.getenv("REMOTE_HOSTNAME")
+REMOTE_PATH = os.getenv("REMOTE_PATH")
+BACKEND_API_URL = os.getenv("BACKEND_API_URL")
+BACKEND_API_KEY= os.getenv("BACKEND_API_KEY")
+
+def load_ssh_config(host_alias):
+	ssh_config = paramiko.SSHConfig()
+	config_path = os.path.expanduser("~/.ssh/config")
+	with open(config_path) as f:
+		ssh_config.parse(f)
+	host_config = ssh_config.lookup(host_alias)
+	hostname = host_config.get("hostname")
+	port = int(host_config.get("port", 22))
+	username = host_config.get("user")
+	keyfile = host_config.get("identityfile", [None])[0]
+
+	if not all([hostname, username, keyfile]):
+		raise ValueError(f"Missing SSH configuration for {host_alias}.")
+
+	return hostname, port, username, keyfile
+
+
+def create_ssh_client(hostname, port, username, keyfile):
+	ssh = paramiko.SSHClient()
+	ssh.load_host_keys(os.path.expanduser("~/.ssh/known_hosts"))
+	ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+	pkey = paramiko.RSAKey.from_private_key_file(keyfile)
+	ssh.connect(hostname=hostname, username=username, port=port, pkey=pkey)
+	return ssh
+
+
+def upload_via_sftp(filename):
+	hostname, port, username, keyfile = load_ssh_config(REMOTE_HOSTNAME)
+
+	transport = paramiko.Transport((hostname, port))
+	pkey = paramiko.RSAKey.from_private_key_file(keyfile)
+	transport.connect(username=username, pkey=pkey)
+	sftp = paramiko.SFTPClient.from_transport(transport)
+
+	remote_file = os.path.join(REMOTE_PATH, os.path.basename(filename))
+	sftp.put(filename, remote_file)
+
+	sftp.close()
+	transport.close()
+
+
+def send_notification_via_ssh(notification_title, notification_info):
+	hostname, port, username, keyfile = load_ssh_config(REMOTE_HOSTNAME)
+
+	with create_ssh_client(hostname, port, username, keyfile) as ssh:
+		data = {
+			"receipent_user_id": 1,
+			"message": {
+				"title": notification_title,
+				"info": notification_info,
+				"category": "mixtapes"
+			}
+		}
+		json_payload = dumps(data)
+
+		# Command reads API key and JSON from stdin
+		notification_cmd = (
+			f"curl -s -X POST '{BACKEND_API_URL}' "
+			f"-H 'Content-Type: application/json' "
+			f"-H 'X-API-Key-Internal: $(head -n1)' "
+			f"-d @-"
+		)
+		stdin, stdout, stderr = ssh.exec_command(notification_cmd)
+		stdin.write(f"{BACKEND_API_KEY}\n{json_payload}")
+		stdin.flush()
+		stdin.channel.shutdown_write()

src/perun/youtube_handler.py

@@ -0,0 +1,74 @@
+import yt_dlp
+import datetime
+import contextlib
+from dotenv import load_dotenv
+import os 
+from helper import return_string_as_html
+
+load_dotenv()
+
+YOUTUBE_CHANNEL_URL = os.getenv("YOUTUBE_CHANNEL_URL")
+
+
+def get_url_for_latest_video():
+	options = {
+	"extract_flat": True,
+	"playlist_items": "1",
+	"quiet": True,
+	"forcejson": True,
+	"simulate": True,
+	}
+	with open(os.devnull, "w") as devnull:
+		with contextlib.redirect_stdout(devnull):
+			with yt_dlp.YoutubeDL(options) as video:
+				info_dict = video.extract_info(YOUTUBE_CHANNEL_URL, download = False)
+	if "entries" in info_dict and len(info_dict["entries"]) > 0:
+		return info_dict["entries"][0]["url"]
+
+def get_youtube_data(url):
+	with yt_dlp.YoutubeDL({"quiet":True,"noprogress":True}) as video:
+		info_dict = video.extract_info(url, download = False)
+	return {"date":datetime.datetime.fromtimestamp(info_dict["timestamp"], datetime.timezone.utc).strftime("%Y-%m-%d"),"title":info_dict["title"],
+			"description":return_string_as_html(info_dict["description"]),"upload_date":info_dict["upload_date"]}
+
+
+def return_download_options(information:dict,track:str)->dict:
+	return {
+		"quiet": True,
+		"noprogress": True,
+		"format": "bestaudio/best", 
+		"extract_audio": True,
+		"audio_format": "mp3",
+		"outtmpl": f"perun-{information['date']}.%(ext)s",
+		"addmetadata": True,
+		"postprocessors":[
+			{"api": "https://sponsor.ajay.app",
+			"categories":{"sponsor"},
+				"key": "SponsorBlock",
+				"when": "after_filter"
+			},
+			{
+				"force_keyframes": False,
+				"key": "ModifyChapters",
+				"remove_chapters_patterns": [],
+				"remove_ranges": [],
+				"remove_sponsor_segments": {"sponsor"},
+				"sponsorblock_chapter_title": "[SponsorBlock]: %(category_names)l"
+			},
+			{
+			"key": "FFmpegExtractAudio",
+			"preferredcodec": "mp3",
+		},
+		{
+		"key": "FFmpegMetadata", 
+		}],
+		"postprocessor_args": [
+			"-metadata", f"title={information['title']}",
+			"-metadata", f"artist=Perun",
+			"-metadata", f"track={track}",
+			"-metadata", f"date={information['date']}",
+			"-metadata", f"comment={information['description']}",
+			"-metadata", f"description={information['description']}",
+		],
+		"merge_output_format": "mp3"  
+	}