A script to subscribe NewPipe's exported channels into any TinyTinyRSS instance
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

111 lines
3.6 KiB

import requests
import json
import pdb
from urllib.parse import urlparse
from getpass import getpass
from bs4 import BeautifulSoup
BASE_FEED_URL = "https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"
TTRSS_API_URL = "https://rss.geekify.de/api/"
def get_channel_feed_url(sub_url):
channel_id = ""
feed_url = ""
url = urlparse(sub_url)
if "/channel" in url.path:
channel_id = url.path.split("/")[-1]
feed_url = BASE_FEED_URL.format(channel_id=channel_id)
elif "/user" in url.path:
req = requests.get(url_str)
soup = BeautifulSoup(req.text, "html.parser")
# we're looking for something like this:
# <link rel="canonical" href="https://www.youtube.com/channel/UC2C_jShtL725hvbm1arSV9w">
channel_link_extracted = soup.find_all("link", rel="canonical")[0].attrs.get("href")
url = urlparse(channel_link_extracted)
channel_id = url.path.split("/")[-1]
feed_url = BASE_FEED_URL.format(channel_id=channel_id)
else:
debug_msg = "Unknown url ({url}) for user \"{name}\""
print(debug_msg.format(url=url_str))
feed_url = ""
return feed_url
class TTRSS():
username = ""
__password = ""
api_endpoint = ""
sid = ""
def __init__(self, api_endpoint="", username="", password=""):
self.username = username
self.api_endpoint = api_endpoint
self.api_level = -1
self.__password = password
self.login()
pass
def login(self):
data = self.__build_request(op="login", user=self.username, password=self.__password)
resp = self.__send_request(data)
self.sid = resp.get("session_id", "")
self.api_level = resp.get("api_level", -1)
if len(self.sid) > 0:
return True
else:
return False
def getCategories(self):
data = self.__build_request(sid=self.sid, op="getCategories", unread_only="f", enable_nested="t", include_empty="f")
resp = self.__send_request(data)
return resp
def subscribeToFeed(self, feed_url, category_id):
data = self.__build_request(sid=self.sid, op="subscribeToFeed", feed_url=feed_url, category_id=category_id)
resp = self.__send_request(data)
return resp
def getFeeds(self, cat_id):
data = self.__build_request(sid=self.sid, op="getFeeds", cat_id=cat_id, unread_only="f")
resp = self.__send_request(data)
return resp
def __build_request(self, *args, **kwargs):
request = kwargs
return request
def __send_request(self, data):
req = requests.post(self.api_endpoint, data=json.dumps(data))
resp = req.json()
if resp.get("status", -1) == 0:
return resp.get("content", {})
else:
print("REQUEST FAILED!")
return {}
print("TinyTinyRSS credentials:")
ttrss = TTRSS(TTRSS_API_URL, input("Username: "), getpass())
cats = ttrss.getCategories()
yt_cat_id_obj = next(filter(lambda x: "YouTube" in x.get("title", ""), cats))
yt_cat_id = yt_cat_id_obj.get("id")
feeds = ttrss.getFeeds(yt_cat_id)
already_subed = []
for feed in feeds:
already_subed.append(feed.get("feed_url"))
with open("subs.json", "r") as f:
newpipe_json = json.loads(f.read())
subs = newpipe_json.get("subscriptions", {})
for sub in subs:
if sub.get("service_id", -1) == 0: # this indicates a youtube link
url_str = sub.get("url", "")
feed_url = get_channel_feed_url(url_str)
if feed_url in already_subed:
print("Already subscribed to: {}".format(sub.get("name")))
else:
print("Subscribing to: {}…".format(sub.get("name")), end="")
resp = ttrss.subscribeToFeed(feed_url, yt_cat_id)
if resp.get("status", {}).get("feed_id", -1) > 0:
print("SUCCESS!")
else:
print("FAILED!")