112 lines
4.9 KiB
Python
112 lines
4.9 KiB
Python
from sqlalchemy.orm import Session
|
|
from xml.etree import ElementTree as ET
|
|
from xml.sax.saxutils import escape
|
|
from email.utils import format_datetime
|
|
from datetime import datetime
|
|
from .database import Program, Episode
|
|
|
|
def get_base_url(request):
|
|
# Retrieve base URL for constructing absolute paths (from FastAPI Request)
|
|
return str(request.base_url).rstrip('/')
|
|
|
|
def generate_master_opml(db: Session, base_url: str, station: str = None):
|
|
opml_outlines = []
|
|
|
|
# Radio 357
|
|
r357 = db.query(Program).filter_by(station="radio357").order_by(Program.name).all() if not station or station == "radio357" else []
|
|
if r357:
|
|
opml_outlines.append(' <outline text="Radio 357">')
|
|
for prog in r357:
|
|
title = escape(prog.name, {'"': """, "'": "'"})
|
|
xmlUrl = f"{base_url}/feeds/radio357/{prog.slug}.xml"
|
|
opml_outlines.append(f' <outline text="{title}" type="rss" xmlUrl="{xmlUrl}" />')
|
|
opml_outlines.append(' </outline>')
|
|
|
|
# RNŚ
|
|
rns = db.query(Program).filter_by(station="rns").order_by(Program.name).all() if not station or station == "rns" else []
|
|
if rns:
|
|
opml_outlines.append(' <outline text="Radio Nowy Świat">')
|
|
for prog in rns:
|
|
title = escape(prog.name, {'"': """, "'": "'"})
|
|
xmlUrl = f"{base_url}/feeds/rns/{prog.slug}.xml"
|
|
opml_outlines.append(f' <outline text="{title}" type="rss" xmlUrl="{xmlUrl}" />')
|
|
opml_outlines.append(' </outline>')
|
|
|
|
# Radio Jazz FM (Direct to source feeds!)
|
|
jazz = db.query(Program).filter_by(station="jazz").order_by(Program.name).all() if not station or station == "jazz" else []
|
|
if jazz:
|
|
opml_outlines.append(' <outline text="Radio Jazz FM">')
|
|
for prog in jazz:
|
|
title = escape(prog.name, {'"': """, "'": "'"})
|
|
# Link directly to their server!
|
|
xmlUrl = f"https://podkasty.radiojazz.fm/@{prog.slug}/feed.xml"
|
|
opml_outlines.append(f' <outline text="{title}" type="rss" xmlUrl="{xmlUrl}" />')
|
|
opml_outlines.append(' </outline>')
|
|
|
|
xml_content = '<?xml version="1.0" encoding="UTF-8"?>\n'
|
|
xml_content += '<opml version="1.0">\n <body>\n'
|
|
xml_content += '\n'.join(opml_outlines) + '\n'
|
|
xml_content += ' </body>\n</opml>'
|
|
return xml_content
|
|
|
|
def generate_podcast_rss(db: Session, station: str, slug: str):
|
|
prog = db.query(Program).filter_by(station=station, slug=slug).first()
|
|
if not prog:
|
|
return None
|
|
|
|
eps = db.query(Episode).filter_by(station=station, program_slug=slug, is_broken=False).filter(Episode.url.isnot(None)).order_by(Episode.pub_date.desc()).all()
|
|
|
|
rss = ET.Element("rss", {"version": "2.0", "xmlns:itunes": "http://www.itunes.com/dtds/podcast-1.0.dtd"})
|
|
channel = ET.SubElement(rss, "channel")
|
|
|
|
ET.SubElement(channel, "title").text = prog.name
|
|
ET.SubElement(channel, "description").text = prog.description or ""
|
|
|
|
if prog.image:
|
|
ET.SubElement(channel, "itunes:image", {"href": prog.image})
|
|
image_el = ET.SubElement(channel, "image")
|
|
ET.SubElement(image_el, "url").text = prog.image
|
|
ET.SubElement(image_el, "title").text = prog.name
|
|
|
|
for ep in eps:
|
|
item = ET.SubElement(channel, "item")
|
|
|
|
date_prefix = f"({ep.pub_date[:10]}) " if ep.pub_date else ""
|
|
ET.SubElement(item, "title").text = f"{date_prefix}{ep.title}"
|
|
ET.SubElement(item, "itunes:author").text = ep.authors or ""
|
|
|
|
if ep.description:
|
|
ET.SubElement(item, "description").text = ep.description
|
|
ET.SubElement(item, "itunes:summary").text = ep.description
|
|
|
|
if ep.pub_date:
|
|
try:
|
|
# radio357 format: 2024-03-22T08:00:00+00:00 or rns: 2024-03-22
|
|
if 'T' in ep.pub_date:
|
|
dt = datetime.fromisoformat(ep.pub_date)
|
|
else:
|
|
dt = datetime.strptime(ep.pub_date, "%Y-%m-%d")
|
|
ET.SubElement(item, "pubDate").text = format_datetime(dt)
|
|
except:
|
|
ET.SubElement(item, "pubDate").text = ep.pub_date
|
|
|
|
if ep.duration_secs:
|
|
ET.SubElement(item, "itunes:duration").text = str(ep.duration_secs)
|
|
|
|
ET.SubElement(item, "guid", {"isPermaLink": "false"}).text = f"{station}-{ep.ep_id}"
|
|
ET.SubElement(item, "enclosure", {"url": ep.url, "type": "audio/mpeg", "length": "0"})
|
|
|
|
if ep.image or prog.image:
|
|
ET.SubElement(item, "itunes:image", {"href": ep.image or prog.image})
|
|
|
|
tree = ET.ElementTree(rss)
|
|
try:
|
|
ET.indent(tree, space=" ", level=0)
|
|
except AttributeError:
|
|
pass # older python versions
|
|
|
|
from io import BytesIO
|
|
f = BytesIO()
|
|
tree.write(f, encoding="utf-8", xml_declaration=True)
|
|
return f.getvalue()
|