API server to manipulate RSS feeds between source and feed reader. Currently allows skipping posts by matching on any fields in a feed.

rsslice.py

#!/usr/bin/env python3

import logging
import re

from flask import Flask, make_response, request
from lxml import etree
from requests import get
from requests.structures import CaseInsensitiveDict


app = Flask(__name__)
app.logger.setLevel(logging.INFO)


@app.get("/")
def help():
    return r"Usage: /skip:title:Sponsor/skip:summary:sponsored%20by/-/https://example.com/feed.xml?category=stuff"


@app.get("/<path:path>")
def route(path: str):
    try:
        ops, url = ("/" + path).split("/-/", 1)
    except ValueError:
        return help(), 400
    pos = url.index("/")
    if url[pos + 1] != "/":
        url = f"{url[:pos]}/{url[pos:]}"
    if request.query_string:
        url = f"{url}?{request.query_string.decode()}"
    app.logger.info("Requesting feed: %s", url)
    headers = CaseInsensitiveDict(request.headers)
    headers.pop("Accept-Encoding", None)
    headers.pop("Host", None)
    call = get(url, headers=headers)
    call.raise_for_status()
    headers = CaseInsensitiveDict(call.headers)
    headers.pop("Content-Encoding", None)
    headers.pop("Content-Length", None)
    headers.pop("Transfer-Encoding", None)
    resp = make_response()
    resp.headers.update(headers)
    if call.status_code != 200:
        app.logger.info("Returning %d response as-is", call.status_code)
        return call.content, call.status_code, headers
    feed = etree.fromstring(call.content)
    if feed.tag.endswith("feed"):
        # Atom
        ns = feed.tag[:-4]
        root = feed
        items = root.findall(f"{ns}entry")
    elif feed.tag.endswith("rss"):
        # RSS
        ns = feed.tag[:-3]
        root = feed.find(f"{ns}channel")
        if root is None:
            return "RSS missing <channel>", 422
        items = root.findall(f"{ns}item")
    else:
        return f"Unknown feed root tag: {feed.tag!r}", 422
    start = len(items)
    for op in ops.split("/")[1:]:
        act, field, value = op.split(":", 2)
        if act != "skip":
            return f"Invalid action: {act!r}", 400
        value = value.lower()
        before = len(items)
        for item in list(items):
            target = (item.findtext(f"{ns}{field}") or "").lower()
            if value.startswith("~"):
                match = re.search(value[1:], target)
            else:
                match = value in target
            if match:
                items.remove(item)
                root.remove(item)
        after = len(items)
        app.logger.debug("Step %r filtered %d item(s)", op, before - after)
    end = len(items)
    app.logger.info("Returning feed with %d->%d items", start, end)
    return etree.tostring(feed), call.status_code, headers


if __name__ == "__main__":
    app.logger.setLevel(logging.DEBUG)
    app.run(debug=True)