rss-bridge-ntfy/webapp.py

"""Flask web application: serves the UI and the REST API used to manage the
bridge (feeds, settings, engine control) entirely from the browser."""

import logging
from datetime import datetime, timezone

import feedparser
from flask import Flask, Response, jsonify, render_template, request

import opml
import store as store_mod
from engine import Engine

logger = logging.getLogger("bridge")


def create_app(store: store_mod.Store, engine: Engine) -> Flask:
    app = Flask(__name__, template_folder="templates", static_folder="static")

    # ---- UI ----------------------------------------------------------------
    @app.get("/")
    def index():
        return render_template("index.html")

    @app.get("/api/health")
    def health():
        return jsonify({"ok": True})

    # ---- status / control --------------------------------------------------
    @app.get("/api/status")
    def status():
        settings = store.get_settings()
        st = dict(engine.status)
        st["enabled"] = settings["enabled"]
        st["sync_interval"] = settings["sync_interval"]
        st["history_count"] = engine.history_count()
        feeds = store.get_feeds()
        st["feed_total"] = len(feeds)
        st["feed_active"] = len([f for f in feeds if f["enabled"]])
        st["topics"] = sorted({f["topic"] for f in feeds if f["topic"]})
        st["server_time"] = datetime.now(timezone.utc).isoformat()
        return jsonify(st)

    @app.post("/api/sync")
    def sync_now():
        engine.trigger_sync()
        return jsonify({"ok": True, "message": "Синхронизация запущена"})

    @app.post("/api/engine")
    def engine_control():
        data = request.get_json(force=True, silent=True) or {}
        action = data.get("action")
        if action == "pause":
            store.update_settings({"enabled": False})
        elif action == "resume":
            store.update_settings({"enabled": True})
            engine.trigger_sync()
        else:
            return jsonify({"ok": False, "error": "unknown action"}), 400
        return jsonify({"ok": True})

    @app.get("/api/logs")
    def logs():
        return jsonify({"logs": list(reversed(engine.ring.records()))})

    @app.post("/api/history/clear")
    def clear_history():
        engine.clear_history()
        return jsonify({"ok": True})

    # ---- settings ----------------------------------------------------------
    @app.get("/api/settings")
    def get_settings():
        return jsonify(store.get_settings())

    @app.put("/api/settings")
    def put_settings():
        data = request.get_json(force=True, silent=True) or {}
        updated = store.update_settings(data)
        engine.trigger_sync()
        return jsonify(updated)

    # ---- feeds -------------------------------------------------------------
    @app.get("/api/feeds")
    def get_feeds():
        return jsonify({"feeds": store.get_feeds()})

    @app.post("/api/feeds")
    def create_feed():
        data = request.get_json(force=True, silent=True) or {}
        if not data.get("url") or not data.get("topic"):
            return jsonify({"ok": False, "error": "url и topic обязательны"}), 400
        feed = store.add_feed(data)
        return jsonify(feed), 201

    @app.put("/api/feeds/<feed_id>")
    def edit_feed(feed_id):
        data = request.get_json(force=True, silent=True) or {}
        feed = store.update_feed(feed_id, data)
        if feed is None:
            return jsonify({"ok": False, "error": "not found"}), 404
        return jsonify(feed)

    @app.delete("/api/feeds/<feed_id>")
    def remove_feed(feed_id):
        ok = store.delete_feed(feed_id)
        return (jsonify({"ok": True}) if ok
                else (jsonify({"ok": False, "error": "not found"}), 404))

    @app.post("/api/feeds/preview")
    def preview_feed():
        """Fetch a feed URL and return a few recent entries (for the UI)."""
        data = request.get_json(force=True, silent=True) or {}
        url = data.get("url", "").strip()
        if not url:
            return jsonify({"ok": False, "error": "url обязателен"}), 400
        settings = store.get_settings()
        parsed = feedparser.parse(url, agent=settings.get("user_agent"))
        entries = [
            {"title": e.get("title", "(no title)"), "link": e.get("link", "")}
            for e in parsed.entries[:5]
        ]
        return jsonify({
            "ok": True,
            "title": parsed.feed.get("title", ""),
            "count": len(parsed.entries),
            "entries": entries,
            "error": str(parsed.get("bozo_exception")) if parsed.bozo else None,
        })

    # ---- OPML import / export ----------------------------------------------
    @app.get("/api/export/opml")
    def export_opml():
        xml = opml.feeds_to_opml(store.get_feeds())
        return Response(
            xml,
            mimetype="text/x-opml",
            headers={"Content-Disposition": "attachment; filename=feeds.opml"},
        )

    @app.post("/api/import/opml")
    def import_opml():
        if "file" in request.files:
            content = request.files["file"].read().decode("utf-8", "replace")
        else:
            content = request.get_data(as_text=True)
        if not content.strip():
            return jsonify({"ok": False, "error": "empty file"}), 400
        try:
            parsed = opml.opml_to_feeds(content)
        except Exception as exc:  # noqa: BLE001
            return jsonify({"ok": False, "error": f"OPML parse error: {exc}"}), 400

        existing = {(f["url"], f["topic"]) for f in store.get_feeds()}
        added = 0
        for feed in parsed:
            if not feed.get("url"):
                continue
            key = (feed["url"], feed["topic"])
            if key in existing:
                continue
            store.add_feed(feed)
            existing.add(key)
            added += 1
        if added:
            engine.trigger_sync()
        logger.info("OPML import: %d new feed(s) of %d", added, len(parsed))
        return jsonify({"ok": True, "added": added, "total": len(parsed)})

    # ---- test notification -------------------------------------------------
    @app.post("/api/test-notify")
    def test_notify():
        data = request.get_json(force=True, silent=True) or {}
        topic = (data.get("topic") or "").strip()
        if not topic:
            return jsonify({"ok": False, "error": "topic обязателен"}), 400
        try:
            engine.send_test(
                topic,
                title=data.get("title") or "rss-bridge-ntfy",
                message=data.get("message") or "Тестовое уведомление ✅",
            )
            return jsonify({"ok": True})
        except Exception as exc:  # noqa: BLE001
            return jsonify({"ok": False, "error": str(exc)}), 502

    return app