sttlab
68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Create Miniflux categories and feeds from feeds.yaml.
|
|
|
|
Env vars required: MINIFLUX_URL, MINIFLUX_USERNAME, MINIFLUX_PASSWORD.
|
|
|
|
Idempotent: existing categories are matched by title, existing feeds are
|
|
matched by feed_url, both fetched from the API before any create call.
|
|
Safe to re-run (e.g. every time the Job runs) — only new entries get
|
|
created.
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
import requests
|
|
import yaml
|
|
|
|
MINIFLUX_URL = os.environ["MINIFLUX_URL"].rstrip("/")
|
|
AUTH = (os.environ["MINIFLUX_USERNAME"], os.environ["MINIFLUX_PASSWORD"])
|
|
FEEDS_FILE = os.path.join(os.path.dirname(__file__), "feeds.yaml")
|
|
|
|
|
|
def api(method, path, **kwargs):
|
|
resp = requests.request(method, f"{MINIFLUX_URL}/v1{path}", auth=AUTH, timeout=60, **kwargs)
|
|
resp.raise_for_status()
|
|
return resp.json() if resp.content else None
|
|
|
|
|
|
def get_or_create_category(name, existing_categories):
|
|
if name in existing_categories:
|
|
return existing_categories[name]
|
|
category = api("POST", "/categories", json={"title": name})
|
|
existing_categories[name] = category["id"]
|
|
print(f"created category: {name}")
|
|
return category["id"]
|
|
|
|
|
|
def main():
|
|
with open(FEEDS_FILE) as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
existing_categories = {c["title"]: c["id"] for c in api("GET", "/categories")}
|
|
existing_feed_urls = {f["feed_url"] for f in api("GET", "/feeds")}
|
|
|
|
for category in data["categories"]:
|
|
category_id = get_or_create_category(category["name"], existing_categories)
|
|
for feed in category["feeds"]:
|
|
if feed["url"] in existing_feed_urls:
|
|
print(f"skip (already added): {feed['title']}")
|
|
continue
|
|
try:
|
|
api(
|
|
"POST",
|
|
"/feeds",
|
|
json={
|
|
"feed_url": feed["url"],
|
|
"category_id": category_id,
|
|
"crawler": feed.get("crawler", False),
|
|
},
|
|
)
|
|
existing_feed_urls.add(feed["url"])
|
|
print(f"added feed: {feed['title']} ({feed['url']})")
|
|
except requests.RequestException as exc:
|
|
print(f"FAILED: {feed['title']} ({feed['url']}): {exc}", file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|