fix typo: mbz is the correct string

This commit is contained in:
Volker
2025-12-27 17:09:31 +01:00
parent c97d22e0f0
commit 6efa6617d3
11 changed files with 8 additions and 8 deletions
+1
View File
@@ -0,0 +1 @@
# mbz_rss_service package
+147
View File
@@ -0,0 +1,147 @@
import shutil
import os
import yaml
from datetime import datetime, timezone
import logging
import uuid
logger = logging.getLogger(__name__)
FEEDS_FILE_PATH = os.path.expandvars(os.environ.get('FEEDS_FILE_PATH', '/var/mbz-rss-feeder/feeds.yml'))
CONFIG_FILE_PATH = os.path.expandvars(os.environ.get('CONFIG_FILE_PATH', '/var/mbz-rss-feeder/etc/mbz-rss-feeder.yml'))
CACHE_DIR = os.path.expandvars(os.environ.get('CACHE_DIR', '/var/mbz-rss-feeder/cache'))
MB_APP_NAME = os.path.expandvars(os.environ.get('MB_APP_NAME', 'mbz-rss-service'))
MB_VERSION = os.path.expandvars(os.environ.get('MB_VERSION', '1'))
MB_CONTACT = os.path.expandvars(os.environ.get('MB_CONTACT', 'someone@somewhere.com'))
# file system dir check
if not os.path.exists(CACHE_DIR):
os.makedirs(CACHE_DIR)
if not os.path.exists(os.path.dirname(FEEDS_FILE_PATH)):
os.makedirs(os.path.dirname(FEEDS_FILE_PATH))
if not os.path.exists(os.path.dirname(CONFIG_FILE_PATH)):
os.makedirs(os.path.dirname(CONFIG_FILE_PATH))
class Config:
def __init__(self):
self._feeds_data = self._load_yaml(FEEDS_FILE_PATH)
if os.path.exists(CONFIG_FILE_PATH):
self._settings = self._load_yaml(CONFIG_FILE_PATH)
else:
self._settings = {
'service': {
'days_back': 0,
'cache_time_hours': 8
}
}
self.FEEDS_FILE_PATH = FEEDS_FILE_PATH
self.CONFIG_FILE_PATH = CONFIG_FILE_PATH
self.CACHE_DIR = CACHE_DIR
self.MB_APP_NAME = MB_APP_NAME
self.MB_VERSION = MB_VERSION
self.MB_CONTACT = MB_CONTACT
try:
with open(os.path.join(os.path.dirname(__file__), '..', 'VERSION')) as f:
self.VERSION = f.read().strip()
except FileNotFoundError:
self.VERSION = '0.0.0'
def __getattr__(self, name):
if name == 'feeds':
return self._feeds_data.get('feeds', [])
if name in self._settings['service']:
return self._settings['service'][name]
raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
def get_settings(self):
return self._settings
def _load_yaml(self, file_path):
if not os.path.exists(file_path):
return {}
with open(file_path, 'r') as f:
return yaml.safe_load(f) or {}
def _save_yaml(self, data, file_path):
if os.path.exists(file_path):
backup_path = f"{file_path}.{datetime.now().strftime('%Y%m%d%H%M%S')}.bak"
shutil.copy(file_path, backup_path)
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, 'w') as f:
yaml.dump(data, f, default_flow_style=False, sort_keys=False)
@property
def feeds(self):
return self._feeds_data.get('feeds', [])
def save_feeds(self):
self._save_yaml(self._feeds_data, FEEDS_FILE_PATH)
def get_feed(self, feed_id):
for feed in self.feeds:
if feed['id'] == feed_id:
return feed
return None
def add_feed(self, name):
now = datetime.now(timezone.utc).isoformat()
new_feed = {
'id': str(uuid.uuid4()),
'name': name,
'artists': [],
'created_at': now,
'updated_at': now
}
if 'feeds' not in self._feeds_data:
self._feeds_data['feeds'] = []
self._feeds_data['feeds'].append(new_feed)
self.save_feeds()
return new_feed
def delete_feed(self, feed_id):
if 'feeds' in self._feeds_data:
self._feeds_data['feeds'] = [feed for feed in self.feeds if feed['id'] != feed_id]
self.save_feeds()
def add_artist_to_feed(self, feed_id, artist_id, artist_name):
for feed in self.feeds:
if feed['id'] == feed_id:
if 'artists' not in feed:
feed['artists'] = []
if not any(a['id'] == artist_id for a in feed['artists']):
feed['artists'].append({'id': artist_id, 'name': artist_name})
feed['updated_at'] = datetime.now(timezone.utc).isoformat()
self.save_feeds()
break
def remove_artist_from_feed(self, feed_id, artist_id):
for feed in self.feeds:
if feed['id'] == feed_id and 'artists' in feed:
original_artist_count = len(feed['artists'])
feed['artists'] = [a for a in feed['artists'] if a['id'] != artist_id]
if len(feed['artists']) < original_artist_count:
feed['updated_at'] = datetime.now(timezone.utc).isoformat()
self.save_feeds()
break
def get_artist_name(self, artist_id):
for feed in self.feeds:
for artist in feed.get('artists', []):
if artist['id'] == artist_id:
return artist['name']
return 'unknown artist'
def save_settings(self, days_back, cache_time_hours):
logger.debug(f"Saving settings: days_back={days_back}, cache_time_hours={cache_time_hours}")
if 'service' not in self._settings:
self._settings['service'] = {}
if days_back is not None:
self._settings['service']['days_back'] = int(days_back)
if cache_time_hours is not None:
self._settings['service']['cache_time_hours'] = int(cache_time_hours)
self._save_yaml(self._settings, CONFIG_FILE_PATH)
# Global instance
config = Config()
+288
View File
@@ -0,0 +1,288 @@
from .config import config
import logging
from flask import Flask, jsonify, render_template, request, redirect, url_for
from datetime import datetime, timedelta, timezone
from email.utils import formatdate, parsedate_to_datetime
import xml.etree.ElementTree as ET
import os
import sys
from . import musicbrainz
log_level_str = os.environ.get('LOG_LEVEL', 'INFO').upper()
log_level = getattr(logging, log_level_str, logging.INFO)
log_file = os.path.expandvars(os.environ.get('LOG_FILE', '/var/mbz-rss-feeder/log/mbz-rss-feeder.log'))
log_format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
mbz_log_level_str = os.environ.get('MBZ_LOG_LEVEL', 'WARNING').upper()
mbz_log_level = getattr(logging, mbz_log_level_str, logging.WARNING)
# file system init
if not os.path.exists(os.path.dirname(log_file)):
os.makedirs(os.path.dirname(log_file))
if not os.path.exists(os.path.dirname(config.CACHE_DIR)):
os.makedirs(os.path.dirname(config.CACHE_DIR))
# Configure root logger
root_logger = logging.getLogger()
root_logger.setLevel(log_level)
formatter = logging.Formatter(log_format)
# configure musicbrainz logger
logging.getLogger('musicbrainzngs').setLevel(mbz_log_level)
# Clear existing handlers
if root_logger.hasHandlers():
root_logger.handlers.clear()
# Add file handler
file_handler = logging.FileHandler(log_file)
file_handler.setFormatter(formatter)
root_logger.addHandler(file_handler)
# Add stdout handler if running with gunicorn (in container)
if 'gunicorn' in sys.argv[0]:
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setFormatter(formatter)
root_logger.addHandler(stdout_handler)
# Get the werkzeug logger and remove its default handlers
werkzeug_logger = logging.getLogger('werkzeug')
for handler in werkzeug_logger.handlers:
werkzeug_logger.removeHandler(handler)
logger = logging.getLogger(__name__)
logger.info(f"Starting mbz-rss-service v{config.VERSION} with the following configuration:")
logger.info(f" - Log Level: {log_level_str}")
logger.info(f" - Log File: {log_file}")
logger.info(f" - Feeds File: {config.FEEDS_FILE_PATH}")
logger.info(f" - Config File: {config.CONFIG_FILE_PATH}")
logger.info(f" - MusicBrainz App Name: {config.MB_APP_NAME}")
logger.info(f" - MusicBrainz App Version: {config.MB_VERSION}")
logger.info(f" - MusicBrainz Contact: {config.MB_CONTACT}")
logger.info(f" - MusicBrainz Log Level: {mbz_log_level_str}")
# init
musicbrainz.init_musicbrainz()
app = Flask(__name__)
class ReverseProxied(object):
def __init__(self, app):
self.app = app
def __call__(self, environ, start_response):
script_name = environ.get('HTTP_X_FORWARDED_PREFIX', '')
if script_name:
logger.debug(f"Redirect prefix: {script_name}")
environ['SCRIPT_NAME'] = script_name
path_info = environ['PATH_INFO']
if path_info.startswith(script_name):
environ['PATH_INFO'] = path_info[len(script_name):]
else:
logger.debug("No redirect prefix detected")
return self.app(environ, start_response)
app.wsgi_app = ReverseProxied(app.wsgi_app)
app.template_folder = 'templates'
def _get_cache_file_path(feed_id):
"""Constructs the full path for a feed's cache file."""
return os.path.join(config.CACHE_DIR, f"{feed_id}.xml")
def _get_cache_last_build_date(cache_file):
"""Parses the cache file and returns its last build date as a timezone-aware datetime object."""
if not os.path.exists(cache_file):
return None
try:
tree = ET.parse(cache_file)
root = tree.getroot()
last_build_date_str = root.findtext('channel/lastBuildDate')
if not last_build_date_str:
return None
last_build_date = parsedate_to_datetime(last_build_date_str)
# Ensure last_build_date is timezone-aware (in UTC)
if last_build_date.tzinfo is None:
return last_build_date.replace(tzinfo=timezone.utc)
return last_build_date
except (ET.ParseError, FileNotFoundError) as e:
logger.warning(f"Could not read or parse cache file {cache_file}: {e}")
return None
def _is_cache_stale(last_build_date, feed_data, cache_time_hours):
"""Determines if the cache is stale based on update times and age."""
if not last_build_date:
return True # No build date means it's stale
# Invalidate if feed config was updated since last cache build
if 'updated_at' in feed_data and feed_data['updated_at']:
feed_updated_at = datetime.fromisoformat(feed_data['updated_at'])
if feed_updated_at > last_build_date:
logger.debug(f"Feed {feed_data['id']} has been updated. Invalidating cache.")
return True
# Invalidate if cache is older than the configured time
if datetime.now(timezone.utc) - last_build_date > timedelta(hours=cache_time_hours):
logger.debug(f"Cache for feed {feed_data['id']} is older than {cache_time_hours} hours. Invalidating.")
return True
return False
def _check_cache(feed_id):
"""Checks for a valid cached feed and returns it if found."""
feed_data = config.get_feed(feed_id)
if not feed_data:
return None # Feed doesn't exist, so no cache.
cache_file = _get_cache_file_path(feed_id)
last_build_date = _get_cache_last_build_date(cache_file)
cache_time_hours = int(config.get_settings().get('service', {}).get('cache_time_hours', 24))
if not _is_cache_stale(last_build_date, feed_data, cache_time_hours):
try:
with open(cache_file, 'r') as f:
logger.debug(f"Serving cached feed for {feed_id}")
return f.read(), {"Content-Type": "application/xml"}
except IOError as e:
logger.warning(f"Could not read cache file {cache_file}: {e}")
return None
@app.route("/")
def index():
logger.debug("Request for index page")
feeds = config.feeds
return render_template('index.html', feeds=feeds)
@app.route("/feed/create", methods=["POST"])
def create_feed():
feed_name = request.form.get('name')
logger.debug(f"Request to create feed with name: {feed_name}")
if feed_name and feed_name.strip():
config.add_feed(feed_name.strip())
return redirect(url_for('index'))
@app.route("/feed/<feed_id>/delete", methods=["POST"])
def delete_feed(feed_id):
logger.debug(f"Request to delete feed with id: {feed_id}")
config.delete_feed(feed_id)
return redirect(url_for('index'))
@app.route("/feed/<feed_id>/edit")
def edit_feed(feed_id):
logger.debug(f"Request to edit feed with id: {feed_id}")
feed = config.get_feed(feed_id)
if not feed:
return "Feed not found", 404
return render_template('feed.html', feed=feed)
@app.route("/feed/<feed_id>/artist/add", methods=["POST"])
def add_artist(feed_id):
artist_id = request.form.get('artist_id')
artist_name = request.form.get('artist_name')
logger.debug(f"Request to add artist '{artist_name}' ({artist_id}) to feed {feed_id}")
if artist_id and artist_name:
config.add_artist_to_feed(feed_id, artist_id, artist_name)
return redirect(url_for('edit_feed', feed_id=feed_id))
@app.route("/feed/<feed_id>/artist/<artist_id>/remove", methods=["POST"])
def remove_artist(feed_id, artist_id):
logger.debug(f"Request to remove artist {artist_id} from feed {feed_id}")
config.remove_artist_from_feed(feed_id, artist_id)
return redirect(url_for('edit_feed', feed_id=feed_id))
@app.route("/artist/search")
def search_artist():
query = request.args.get('q', '')
logger.debug(f"Request to search for artist with query: '{query}'")
artists = musicbrainz.search_artists(query)
return jsonify(artists)
@app.route("/settings", methods=["GET", "POST"])
def settings():
if request.method == "POST":
logger.debug("Request to update settings")
days_back = request.form.get('days_back')
cache_time_hours = request.form.get('cache_time_hours')
config.save_settings(days_back, cache_time_hours)
return redirect(url_for('settings'))
logger.debug("Request for settings page")
return render_template('settings.html', settings=config)
@app.route("/opml")
def opml():
logger.debug("Request for OPML file")
feeds = config.feeds
opml_content = render_template('opml.xml', feeds=feeds)
return opml_content, {"Content-Type": "application/xml"}
@app.route('/feed/<feed_id>')
def get_feed_rss(feed_id):
logger.debug(f"Request for RSS feed with id: {feed_id}")
# Check cache for a valid feed before generating it
cached_response = _check_cache(feed_id)
if cached_response:
return cached_response
# cache outdated or not found, generate a new feed
logger.debug(f"Generating new feed for {feed_id}")
feed_data = config.get_feed(feed_id)
if not feed_data:
return "Feed not found", 404
# Add rfc822 formatted updated_at for template
if 'updated_at' in feed_data and feed_data['updated_at']:
updated_at_dt = datetime.fromisoformat(feed_data['updated_at'])
feed_data['updated_at_rfc822'] = formatdate(updated_at_dt.timestamp())
else:
# Fallback for older feeds without updated_at
now_utc = datetime.now(timezone.utc)
feed_data['updated_at_rfc822'] = formatdate(now_utc.timestamp())
all_releases = []
if 'artists' in feed_data:
for artist in feed_data['artists']:
releases = musicbrainz.get_artist_releases(artist['id'])
all_releases.extend(releases)
# Sort releases by date, newest first
all_releases.sort(key=lambda r: r.get('date', '0000-00-00'), reverse=True)
last_build_date = formatdate(datetime.now(timezone.utc).timestamp())
rss_content = render_template('feed.xml', feed=feed_data, releases=all_releases, last_build_date=last_build_date)
cache_dir = config.CACHE_DIR
os.makedirs(cache_dir, exist_ok=True)
cache_file = os.path.join(cache_dir, f"{feed_id}.xml")
try:
with open(cache_file, 'w') as f:
f.write(rss_content)
logger.debug(f"Cached feed '{feed_data['name']}' at {cache_file}")
except IOError as e:
logger.warning(f"Could not write feed '{feed_data['name']}' to cache file {cache_file}: {e}")
return rss_content, {"Content-Type": "application/xml"}
@app.route("/health")
def health_check():
logger.debug("Health check requested")
return jsonify({"status": "ok"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8080, debug=True)
+104
View File
@@ -0,0 +1,104 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>MusicBrainz RSS Feeder</title>
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<style>
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif;
line-height: 1.6;
color: #333;
background-color: #fdfdfd;
margin: 0;
}
.container {
max-width: 800px;
margin: 0 auto;
padding: 20px;
}
header {
display: flex;
justify-content: space-between;
align-items: center;
padding-bottom: 20px;
border-bottom: 1px solid #eee;
margin-bottom: 20px;
}
header h1 {
font-size: 24px;
margin: 0;
}
header h1 a {
text-decoration: none;
color: #333;
}
nav a {
margin-left: 20px;
text-decoration: none;
color: #555;
}
h2, h3 {
border-bottom: 1px solid #eee;
padding-bottom: 10px;
margin-top: 40px;
}
ul {
list-style-type: none;
padding: 0;
}
li {
padding: 10px 0;
border-bottom: 1px solid #eee;
display: flex;
justify-content: space-between;
align-items: center;
}
li:last-child {
border-bottom: none;
}
form {
margin-top: 20px;
}
input[type="text"] {
width: 100%;
padding: 8px;
box-sizing: border-box;
border: 1px solid #ccc;
border-radius: 4px;
}
button {
background-color: #333;
color: white;
padding: 8px 15px;
border: none;
border-radius: 4px;
cursor: pointer;
}
button:hover {
background-color: #555;
}
.form-inline {
display: inline;
margin-top: 0;
}
.item-actions {
display: flex;
align-items: center;
gap: 10px;
}
</style>
</head>
<body>
<div class="container">
<header>
<h1><a href="{{ url_for('index') }}">MusicBrainz RSS Feeder</a></h1>
<nav>
<a href="{{ url_for('index') }}">Feeds</a>
<a href="{{ url_for('settings') }}">Settings</a>
</nav>
</header>
{% block content %}{% endblock %}
</div>
</body>
</html>
+31
View File
@@ -0,0 +1,31 @@
{% extends "base.html" %}
{% block content %}
<h2>Feeds</h2>
<ul>
{% for feed in feeds %}
<li>
<div>
<a href="{{ url_for('edit_feed', feed_id=feed.id) }}">{{ feed.name }}</a>
({{ feed.artists|length }})
</div>
<div class="item-actions">
<a href="{{ url_for('get_feed_rss', feed_id=feed.id) }}">RSS</a>
<form class="form-inline" action="{{ url_for('delete_feed', feed_id=feed.id) }}" method="post">
<button type="submit">Delete</button>
</form>
</div>
</li>
{% else %}
<li>No feeds configured.</li>
{% endfor %}
</ul>
<h3>Create New Feed</h3>
<form action="{{ url_for('create_feed') }}" method="post">
<label for="name">Feed Name:</label>
<input type="text" id="name" name="name" required>
<button type="submit">Create</button>
</form>
{% endblock %}
+13
View File
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<opml version="2.0">
<head>
<title>MusicBrainz Artist Feeds</title>
</head>
<body>
<outline text="MusicBrainz Artist Feeds" title="MusicBrainz Artist Feeds">
{% for feed in feeds %}
<outline type="rss" text="{{ (feed.artists | map(attribute='name') | sort | join(', ')) }}" title="{{ feed.name }} ({{ feed.artists | length }} artists)" xmlUrl="{{ url_for('get_feed_rss', feed_id=feed.id, _external=True) }}"/>
{% endfor %}
</outline>
</body>
</opml>
+19
View File
@@ -0,0 +1,19 @@
{% extends "base.html" %}
{% block content %}
<h2>Settings</h2>
<form method="post">
<div>
<label for="days_back">Days to look back for releases:</label>
<input type="number" id="days_back" name="days_back" value="{{ settings.days_back }}" required>
</div>
<br>
<div>
<label for="cache_time_hours">Cache time (hours):</label>
<input type="number" id="cache_time_hours" name="cache_time_hours" value="{{ settings.cache_time_hours }}" required>
</div>
<br>
<button type="submit">Save Settings</button>
</form>
{% endblock %}
+7
View File
@@ -0,0 +1,7 @@
Flask
PyYAML
musicbrainzngs
feedgen
gunicorn
pytest
pytz