288 lines
10 KiB
Python
288 lines
10 KiB
Python
# ReadFeeder
|
||
# Parse newsfeed, generate audiofiles for it and post them to the fediverse
|
||
|
||
import os
|
||
import hashlib
|
||
import regex
|
||
|
||
import feedparser
|
||
from mastodon import Mastodon
|
||
|
||
from langdetect import detect
|
||
|
||
from pydub import AudioSegment
|
||
import eyed3
|
||
from gtts import gTTS
|
||
import elevenlabs
|
||
|
||
import config
|
||
|
||
## Initialization #############################################################
|
||
|
||
# Match whitespace
|
||
regex_whitespace = regex.compile(r"\s")
|
||
|
||
# Match speakable Text (at least one char in any language)
|
||
regex_speakable = regex.compile(r'[\d\p{L}]+')
|
||
|
||
# Pfade
|
||
base_path = os.path.dirname(os.path.realpath(__file__))
|
||
post_directory = base_path + "/posts"
|
||
sentence_directory = base_path + "/sentences"
|
||
|
||
# Load configuration
|
||
configuration = config.Config(base_path + "/config.yml")
|
||
# Just flat config, no checks
|
||
cfg = configuration.config
|
||
|
||
# Create a Mastodon client
|
||
mastodon_client = Mastodon(
|
||
access_token=cfg["mastodon"]["access_token"],
|
||
api_base_url=cfg["mastodon"]["instance_url"]
|
||
)
|
||
|
||
if cfg["elevenlabs"]["enabled"]:
|
||
# Login to ElevenLabs
|
||
elevenlabs.set_api_key(cfg["elevenlabs"]["api_key"])
|
||
|
||
## Functions ##################################################################
|
||
|
||
# Function to create a new post on mastodon
|
||
def mastodon_post(text, text_hash, entry, audio_file_path):
|
||
|
||
# Upload audio file
|
||
media_dict = mastodon_client.media_post(
|
||
audio_file_path,
|
||
mime_type="audio/mpeg",
|
||
description="Audio-Post: " + text
|
||
)
|
||
|
||
# Create the post
|
||
status = ""
|
||
if cfg["mastodon"]["post"]["title"]:
|
||
status += entry.title + " – "
|
||
if cfg["mastodon"]["post"]["text"]:
|
||
status += text
|
||
if cfg["mastodon"]["post"]["date"]:
|
||
status += " – " + entry.pubdate
|
||
if cfg["mastodon"]["post"]["link"]:
|
||
status += " – Quelle: " + entry.link
|
||
|
||
post_params = {
|
||
"status": status,
|
||
"visibility": cfg["mastodon"]["visibility"],
|
||
"media_ids": [media_dict["id"]]
|
||
}
|
||
|
||
mastodon_client.status_post(**post_params)
|
||
|
||
print(f"Post for '{text_hash}' created successfully.")
|
||
|
||
# Function to generate a hash for content
|
||
def generate_hash(content):
|
||
return hashlib.md5(regex_whitespace.sub("",content).lower().encode()).hexdigest()
|
||
|
||
# Function to generate a hash for content
|
||
def generate_old_hash(content):
|
||
return hashlib.md5(content.lower().encode()).hexdigest()
|
||
|
||
# Function to generate a silent MP3 file of a given duration in ms
|
||
def generate_silent_mp3(filename, duration=20):
|
||
silence = AudioSegment.silent(duration=duration)
|
||
silence.export(filename, format="mp3")
|
||
|
||
# Function to generate mp3 file from text using gTTS
|
||
def generate_mp3_from_text(text, link, filename, lang):
|
||
try:
|
||
# If there is not at least one speakable character...
|
||
if not regex_speakable.search(text):
|
||
raise AssertionError ("unspeakable")
|
||
|
||
# Use elevenlabs or gTTS?
|
||
if cfg["elevenlabs"]["enabled"]:
|
||
# Use elevenlabs
|
||
audio = elevenlabs.generate(
|
||
text=text,
|
||
voice=elevenlabs.Voice(
|
||
voice_id=cfg["elevenlabs"]["voice"]["id"],
|
||
settings=elevenlabs.VoiceSettings(
|
||
stability=cfg["elevenlabs"]["voice"]["stability"],
|
||
similarity_boost=cfg["elevenlabs"]["voice"]["similarity_boost"],
|
||
style=cfg["elevenlabs"]["voice"]["style"],
|
||
use_speaker_boost=cfg["elevenlabs"]["voice"]["use_speaker_boost"]
|
||
)
|
||
),
|
||
model='eleven_multilingual_v2'
|
||
)
|
||
elevenlabs.save(audio, filename)
|
||
else:
|
||
# Use gTTS
|
||
tts = gTTS(text, lang=lang)
|
||
tts.save(filename)
|
||
|
||
except AssertionError:
|
||
# Generate a silent MP3 file when there is an assertion error
|
||
generate_silent_mp3(filename)
|
||
except ValueError:
|
||
# Unsupported Language in gTTS
|
||
tts = gTTS(text, lang=cfg["languages"]["default"])
|
||
tts.save(filename)
|
||
|
||
|
||
# Add ID3 tags to the file using eyed3
|
||
audiofile = eyed3.load(filename)
|
||
|
||
if not audiofile.tag:
|
||
audiofile.initTag()
|
||
|
||
audiofile.tag.comments.set(text)
|
||
# Add some more meta data
|
||
audiofile.tag.title = text[0:24] # Add a shortend form of the text as title
|
||
audiofile.tag.artist = cfg["audio"]["artist"] # Add the artist
|
||
audiofile.tag.lyrics.set(text) # Add the full content as lyrics
|
||
audiofile.tag.genre = 101 # Add speech as a genre
|
||
audiofile.tag.audio_source_url = link
|
||
audiofile.tag.save()
|
||
|
||
# Create a text file with the same name as the mp3 file
|
||
write_text_file(text, filename.replace("mp3","txt"))
|
||
|
||
|
||
# Function to split content into sentences while keeping separators at the end
|
||
def split_into_sentences(content):
|
||
# The more separators are used, the more parts can be reused, but if
|
||
# sentences are to small, the quality will be very low.
|
||
|
||
# i.e. you could use " " as separator, effictively splitting into single
|
||
# words. So each possible word will only be generated oncem but the
|
||
# consisten tonality of sentences might be lost and the resulting
|
||
# combined file will sound awful.
|
||
separators = '.!?…'
|
||
sentences = []
|
||
sentence = ""
|
||
for char in content:
|
||
sentence += char
|
||
if char in separators:
|
||
# If there is at least one speakable character...
|
||
if regex_speakable.search(sentence):
|
||
# Append sentence
|
||
sentences.append(sentence)
|
||
# Reset sentence
|
||
sentence = ""
|
||
return sentences
|
||
|
||
# Function to generate a text file with the given content and hash as the name
|
||
def write_text_file(content, filename):
|
||
with open(filename, 'w', encoding='utf-8') as text_file:
|
||
text_file.write(content) # Write the full content to the text file
|
||
|
||
# Post-generation hook function to post the result of the generation process
|
||
def join_mp3(content, content_hash, sentence_files, link):
|
||
# Filename
|
||
audio_filename = os.path.join(post_directory, content_hash + '.mp3')
|
||
|
||
# Join all sentence MP3 files into one larger MP3 file for the current entry
|
||
# Additionaly write a playlist with all segments.
|
||
combined_audio = AudioSegment.empty()
|
||
with open(audio_filename.replace("mp3","m3u"), 'w', encoding='utf-8') as pl_file:
|
||
for sentence_file in sentence_files:
|
||
audio = AudioSegment.from_mp3(sentence_file)
|
||
combined_audio += audio
|
||
pl_file.write(os.path.relpath(sentence_file, post_directory) + "\n")
|
||
pl_file.close()
|
||
|
||
# Save the combined audio
|
||
combined_audio.export(audio_filename, format="mp3")
|
||
|
||
# Add a COMMENT (COMM) tag with the full content of the entry
|
||
audiofile = eyed3.load(audio_filename)
|
||
audiofile.tag.artist = cfg["audio"]["artist"] # Add the artist
|
||
audiofile.tag.title = content[0:24] # Add a shortend form of the text as title
|
||
audiofile.tag.comments.set(content) # Add the full content as a comment
|
||
audiofile.tag.lyrics.set(content) # Add the full content as lyrics
|
||
audiofile.tag.genre = 101 # Add speech as a genre
|
||
audiofile.tag.audio_source_url = link # Add the source url
|
||
audiofile.tag.save()
|
||
|
||
# Create a text file with the same name as the mp3 file
|
||
write_text_file(content, audio_filename.replace("mp3","txt"))
|
||
|
||
# Function to process RSS feed
|
||
def process_rss_feed(feed_url, post_directory, sentence_directory):
|
||
feed = feedparser.parse(feed_url)
|
||
|
||
for entry in feed.entries:
|
||
|
||
# Load content
|
||
if cfg["rss"]["use_field"] == "title":
|
||
content = entry.title
|
||
elif cfg["rss"]["use_field"] == "description":
|
||
content = entry.description
|
||
else:
|
||
content = "No content."
|
||
|
||
# Filter words
|
||
for old_str, new_str in cfg["word_filter"].items():
|
||
content = content.replace(old_str, new_str)
|
||
|
||
# RegEx filter
|
||
for old_str, new_str in cfg["regex_filter"].items():
|
||
regex_filter = regex.compile(old_str)
|
||
content = regex_filter.sub(new_str, content)
|
||
|
||
# Generate a hash for the content
|
||
content_hash = generate_hash(content)
|
||
|
||
# Check for an MP3 file in the 'posts' directory
|
||
post_mp3_file = os.path.join(post_directory, content_hash + '.mp3')
|
||
|
||
if not os.path.exists(post_mp3_file):
|
||
# Split content into sentences and process each sentence
|
||
sentences = split_into_sentences(content)
|
||
|
||
sentence_files = []
|
||
|
||
for sentence in sentences:
|
||
|
||
# Detect language of the sentence
|
||
try:
|
||
sentence_language = detect(sentence)
|
||
except:
|
||
sentence_language = cfg["languages"]["default"]
|
||
|
||
if sentence_language not in cfg["languages"]["used"]:
|
||
sentence_language = cfg["languages"]["default"]
|
||
|
||
# Generate Hash for the sentence
|
||
sentence_hash = generate_hash(sentence)
|
||
|
||
# Check for an MP3 file in the 'sentences' directory
|
||
sentence_mp3_file = os.path.join(sentence_directory, sentence_hash + '.mp3')
|
||
|
||
if not os.path.exists(sentence_mp3_file):
|
||
# Generate MP3 file using gTTS with a German voice
|
||
generate_mp3_from_text(sentence, entry.link, sentence_mp3_file, sentence_language)
|
||
|
||
sentence_files.append(sentence_mp3_file)
|
||
|
||
# Join all sentence MP3 files into one larger MP3 file for the current entry
|
||
join_mp3(content, content_hash, sentence_files, entry.link)
|
||
|
||
# Post on Mastodon
|
||
if cfg["mastodon"]["enabled"]:
|
||
mastodon_post(content, content_hash, entry, post_mp3_file)
|
||
|
||
print(f"Audiofiles for '{content_hash}' generated successfully.")
|
||
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# If executed standalone
|
||
|
||
if not os.path.exists(post_directory):
|
||
os.makedirs(post_directory)
|
||
|
||
if not os.path.exists(sentence_directory):
|
||
os.makedirs(sentence_directory)
|
||
|
||
process_rss_feed(cfg["rss"]["feed_url"], post_directory, sentence_directory)
|